http://git-wip-us.apache.org/repos/asf/nifi/blob/034ee6de/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardProcessorNode.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardProcessorNode.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardProcessorNode.java index cbd0f88..0c39eda 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardProcessorNode.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardProcessorNode.java @@ -16,14 +16,6 @@ */ package org.apache.nifi.controller; -import org.apache.nifi.annotation.behavior.EventDriven; -import org.apache.nifi.annotation.behavior.SideEffectFree; -import org.apache.nifi.annotation.behavior.SupportsBatching; -import org.apache.nifi.annotation.behavior.TriggerSerially; -import org.apache.nifi.annotation.behavior.TriggerWhenAnyDestinationAvailable; -import org.apache.nifi.annotation.behavior.TriggerWhenEmpty; -import org.apache.nifi.annotation.documentation.CapabilityDescription; - import static java.util.Objects.requireNonNull; import java.util.ArrayList; @@ -43,6 +35,17 @@ import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantReadWriteLock; +import org.apache.commons.lang3.builder.EqualsBuilder; +import org.apache.commons.lang3.builder.HashCodeBuilder; +import org.apache.nifi.annotation.behavior.EventDriven; +import org.apache.nifi.annotation.behavior.InputRequirement; +import org.apache.nifi.annotation.behavior.InputRequirement.Requirement; +import org.apache.nifi.annotation.behavior.SideEffectFree; +import org.apache.nifi.annotation.behavior.SupportsBatching; +import org.apache.nifi.annotation.behavior.TriggerSerially; +import org.apache.nifi.annotation.behavior.TriggerWhenAnyDestinationAvailable; +import org.apache.nifi.annotation.behavior.TriggerWhenEmpty; +import org.apache.nifi.annotation.documentation.CapabilityDescription; import org.apache.nifi.components.ValidationContext; import org.apache.nifi.components.ValidationResult; import org.apache.nifi.connectable.Connectable; @@ -61,8 +64,6 @@ import org.apache.nifi.processor.Processor; import org.apache.nifi.processor.Relationship; import org.apache.nifi.scheduling.SchedulingStrategy; import org.apache.nifi.util.FormatUtils; -import org.apache.commons.lang3.builder.EqualsBuilder; -import org.apache.commons.lang3.builder.HashCodeBuilder; import org.quartz.CronExpression; import org.slf4j.LoggerFactory; @@ -73,1185 +74,1242 @@ import org.slf4j.LoggerFactory; */ public class StandardProcessorNode extends ProcessorNode implements Connectable { - public static final String BULLETIN_OBSERVER_ID = "bulletin-observer"; - - public static final TimeUnit DEFAULT_TIME_UNIT = TimeUnit.MILLISECONDS; - public static final String DEFAULT_YIELD_PERIOD = "1 sec"; - public static final String DEFAULT_PENALIZATION_PERIOD = "30 sec"; - private final AtomicReference<ProcessGroup> processGroup; - private final Processor processor; - private final AtomicReference<String> identifier; - private final Map<Connection, Connectable> destinations; - private final Map<Relationship, Set<Connection>> connections; - private final AtomicReference<Set<Relationship>> undefinedRelationshipsToTerminate; - private final AtomicReference<List<Connection>> incomingConnectionsRef; - private final ReentrantReadWriteLock rwLock; - private final Lock readLock; - private final Lock writeLock; - private final AtomicBoolean isolated; - private final AtomicBoolean lossTolerant; - private final AtomicReference<ScheduledState> scheduledState; - private final AtomicReference<String> comments; - private final AtomicReference<String> name; - private final AtomicReference<Position> position; - private final AtomicReference<String> annotationData; - private final AtomicReference<String> schedulingPeriod; // stored as string so it's presented to user as they entered it - private final AtomicReference<String> yieldPeriod; - private final AtomicReference<String> penalizationPeriod; - private final AtomicReference<Map<String, String>> style; - private final AtomicInteger concurrentTaskCount; - private final AtomicLong yieldExpiration; - private final AtomicLong schedulingNanos; - private final boolean triggerWhenEmpty; - private final boolean sideEffectFree; - private final boolean triggeredSerially; - private final boolean triggerWhenAnyDestinationAvailable; - private final boolean eventDrivenSupported; - private final boolean batchSupported; - private final ValidationContextFactory validationContextFactory; - private final ProcessScheduler processScheduler; - private long runNanos = 0L; - - private SchedulingStrategy schedulingStrategy; // guarded by read/write lock - - @SuppressWarnings("deprecation") - public StandardProcessorNode(final Processor processor, final String uuid, final ValidationContextFactory validationContextFactory, - final ProcessScheduler scheduler, final ControllerServiceProvider controllerServiceProvider) { - super(processor, uuid, validationContextFactory, controllerServiceProvider); - - this.processor = processor; - identifier = new AtomicReference<>(uuid); - destinations = new HashMap<>(); - connections = new HashMap<>(); - incomingConnectionsRef = new AtomicReference<List<Connection>>(new ArrayList<Connection>()); - scheduledState = new AtomicReference<>(ScheduledState.STOPPED); - rwLock = new ReentrantReadWriteLock(false); - readLock = rwLock.readLock(); - writeLock = rwLock.writeLock(); - lossTolerant = new AtomicBoolean(false); - final Set<Relationship> emptySetOfRelationships = new HashSet<>(); - undefinedRelationshipsToTerminate = new AtomicReference<>(emptySetOfRelationships); - comments = new AtomicReference<>(""); - name = new AtomicReference<>(processor.getClass().getSimpleName()); - schedulingPeriod = new AtomicReference<>("0 sec"); - schedulingNanos = new AtomicLong(MINIMUM_SCHEDULING_NANOS); - yieldPeriod = new AtomicReference<>(DEFAULT_YIELD_PERIOD); - yieldExpiration = new AtomicLong(0L); - concurrentTaskCount = new AtomicInteger(1); - position = new AtomicReference<>(new Position(0D, 0D)); - style = new AtomicReference<>(Collections.unmodifiableMap(new HashMap<String, String>())); - this.processGroup = new AtomicReference<>(); - processScheduler = scheduler; - annotationData = new AtomicReference<>(); - isolated = new AtomicBoolean(false); - penalizationPeriod = new AtomicReference<>(DEFAULT_PENALIZATION_PERIOD); - - final Class<?> procClass = processor.getClass(); - triggerWhenEmpty = procClass.isAnnotationPresent(TriggerWhenEmpty.class) || procClass.isAnnotationPresent(org.apache.nifi.processor.annotation.TriggerWhenEmpty.class); - sideEffectFree = procClass.isAnnotationPresent(SideEffectFree.class) || procClass.isAnnotationPresent(org.apache.nifi.processor.annotation.SideEffectFree.class); - batchSupported = procClass.isAnnotationPresent(SupportsBatching.class) || procClass.isAnnotationPresent(org.apache.nifi.processor.annotation.SupportsBatching.class); - triggeredSerially = procClass.isAnnotationPresent(TriggerSerially.class) || procClass.isAnnotationPresent(org.apache.nifi.processor.annotation.TriggerSerially.class); - triggerWhenAnyDestinationAvailable = procClass.isAnnotationPresent(TriggerWhenAnyDestinationAvailable.class) - || procClass.isAnnotationPresent(org.apache.nifi.processor.annotation.TriggerWhenAnyDestinationAvailable.class); - this.validationContextFactory = validationContextFactory; - eventDrivenSupported = (procClass.isAnnotationPresent(EventDriven.class) - || procClass.isAnnotationPresent(org.apache.nifi.processor.annotation.EventDriven.class)) && !triggeredSerially && !triggerWhenEmpty; - schedulingStrategy = SchedulingStrategy.TIMER_DRIVEN; - } - - /** - * @return comments about this specific processor instance - */ - @Override - public String getComments() { - return comments.get(); - } - - /** - * Provides and opportunity to retain information about this particular processor instance - * - * @param comments new comments - */ - @Override - public void setComments(final String comments) { - writeLock.lock(); - try { - if (isRunning()) { - throw new IllegalStateException("Cannot modify Processor configuration while the Processor is running"); - } - this.comments.set(comments); - } finally { - writeLock.unlock(); - } - } - - @Override - public ScheduledState getScheduledState() { - return scheduledState.get(); - } - - @Override - public Position getPosition() { - return position.get(); - } - - @Override - public void setPosition(Position position) { - this.position.set(position); - } - - @Override - public Map<String, String> getStyle() { - return style.get(); - } - - @Override - public void setStyle(final Map<String, String> style) { - if (style != null) { - this.style.set(Collections.unmodifiableMap(new HashMap<>(style))); - } - } - - @Override - public String getIdentifier() { - return identifier.get(); - } - - /** - * @return if true flow file content generated by this processor is considered loss tolerant - */ - @Override - public boolean isLossTolerant() { - return lossTolerant.get(); - } - - @Override - public boolean isIsolated() { - return isolated.get(); - } - - /** - * @return true if the processor has the {@link TriggerWhenEmpty} annotation, false otherwise. - */ - @Override - public boolean isTriggerWhenEmpty() { - return triggerWhenEmpty; - } - - /** - * @return true if the processor has the {@link SideEffectFree} annotation, false otherwise. - */ - @Override - public boolean isSideEffectFree() { - return sideEffectFree; - } - - @Override - public boolean isHighThroughputSupported() { - return batchSupported; - } - - /** - * @return true if the processor has the {@link TriggerWhenAnyDestinationAvailable} annotation, false otherwise. - */ - @Override - public boolean isTriggerWhenAnyDestinationAvailable() { - return triggerWhenAnyDestinationAvailable; - } - - /** - * Indicates whether flow file content made by this processor must be persisted - * - * @param lossTolerant tolerant - */ - @Override - public void setLossTolerant(final boolean lossTolerant) { - writeLock.lock(); - try { - if (isRunning()) { - throw new IllegalStateException("Cannot modify Processor configuration while the Processor is running"); - } - this.lossTolerant.set(lossTolerant); - } finally { - writeLock.unlock(); - } - } - - /** - * Indicates whether the processor runs on only the primary node. - * - * @param isolated isolated - */ - public void setIsolated(final boolean isolated) { - writeLock.lock(); - try { - if (isRunning()) { - throw new IllegalStateException("Cannot modify Processor configuration while the Processor is running"); - } - this.isolated.set(isolated); - } finally { - writeLock.unlock(); - } - } - - @Override - public boolean isAutoTerminated(final Relationship relationship) { - final Set<Relationship> terminatable = undefinedRelationshipsToTerminate.get(); - if (terminatable == null) { - return false; - } - return terminatable.contains(relationship); - } - - @Override - public void setAutoTerminatedRelationships(final Set<Relationship> terminate) { - writeLock.lock(); - try { - if (isRunning()) { - throw new IllegalStateException("Cannot modify Processor configuration while the Processor is running"); - } - - for (final Relationship rel : terminate) { - if (!getConnections(rel).isEmpty()) { - throw new IllegalStateException("Cannot mark relationship '" + rel.getName() + "' as auto-terminated because Connection already exists with this relationship"); - } - } - undefinedRelationshipsToTerminate.set(new HashSet<>(terminate)); - } finally { - writeLock.unlock(); - } - } - - /** - * @return an unmodifiable Set that contains all of the ProcessorRelationship objects that are configured to be auto-terminated - */ - @Override - public Set<Relationship> getAutoTerminatedRelationships() { - Set<Relationship> relationships = undefinedRelationshipsToTerminate.get(); - if (relationships == null) { - relationships = new HashSet<>(); - } - return Collections.unmodifiableSet(relationships); - } - - @Override - public String getName() { - return name.get(); - } - - /** - * @return the value of the processor's {@link CapabilityDescription} annotation, if one exists, else <code>null</code>. - */ - @SuppressWarnings("deprecation") - public String getProcessorDescription() { - CapabilityDescription capDesc = processor.getClass().getAnnotation(CapabilityDescription.class); - String description = null; - if (capDesc != null) { - description = capDesc.value(); - } else { - final org.apache.nifi.processor.annotation.CapabilityDescription deprecatedCapDesc - = processor.getClass().getAnnotation(org.apache.nifi.processor.annotation.CapabilityDescription.class); - if (deprecatedCapDesc != null) { - description = deprecatedCapDesc.value(); - } - } - - return description; - } - - @Override - public void setName(final String name) { - writeLock.lock(); - try { - if (isRunning()) { - throw new IllegalStateException("Cannot modify Processor configuration while the Processor is running"); - } - this.name.set(name); - } finally { - writeLock.unlock(); - } - } - - /** - * @param timeUnit determines the unit of time to represent the scheduling period. If null will be reported in units of {@link #DEFAULT_SCHEDULING_TIME_UNIT} - * @return the schedule period that should elapse before subsequent cycles of this processor's tasks - */ - @Override - public long getSchedulingPeriod(final TimeUnit timeUnit) { - return timeUnit.convert(schedulingNanos.get(), TimeUnit.NANOSECONDS); - } - - @Override - public boolean isEventDrivenSupported() { - readLock.lock(); - try { - return this.eventDrivenSupported; - } finally { - readLock.unlock(); - } - } - - /** - * Updates the Scheduling Strategy used for this Processor - * - * @param schedulingStrategy strategy - * - * @throws IllegalArgumentException if the SchedulingStrategy is not not applicable for this Processor - */ - @Override - public void setSchedulingStrategy(final SchedulingStrategy schedulingStrategy) { - writeLock.lock(); - try { - if (schedulingStrategy == SchedulingStrategy.EVENT_DRIVEN && !eventDrivenSupported) { - // not valid. Just ignore it. We don't throw an Exception because if a developer changes a Processor so that - // it no longer supports EventDriven mode, we don't want the app to fail to startup if it was already in Event-Driven - // Mode. Instead, we will simply leave it in Timer-Driven mode - return; - } - - this.schedulingStrategy = schedulingStrategy; - setIsolated(schedulingStrategy == SchedulingStrategy.PRIMARY_NODE_ONLY); - } finally { - writeLock.unlock(); - } - } - - /** - * @return the currently configured scheduling strategy - */ - @Override - public SchedulingStrategy getSchedulingStrategy() { - readLock.lock(); - try { - return this.schedulingStrategy; - } finally { - readLock.unlock(); - } - } - - @Override - public String getSchedulingPeriod() { - return schedulingPeriod.get(); - } - - @Override - public void setScheduldingPeriod(final String schedulingPeriod) { - writeLock.lock(); - try { - if (isRunning()) { - throw new IllegalStateException("Cannot modify Processor configuration while the Processor is running"); - } - - switch (schedulingStrategy) { - case CRON_DRIVEN: { - try { - new CronExpression(schedulingPeriod); - } catch (final Exception e) { - throw new IllegalArgumentException("Scheduling Period is not a valid cron expression: " + schedulingPeriod); - } - } - break; - case PRIMARY_NODE_ONLY: - case TIMER_DRIVEN: { - final long schedulingNanos = FormatUtils.getTimeDuration(requireNonNull(schedulingPeriod), TimeUnit.NANOSECONDS); - if (schedulingNanos < 0) { - throw new IllegalArgumentException("Scheduling Period must be positive"); - } - this.schedulingNanos.set(Math.max(MINIMUM_SCHEDULING_NANOS, schedulingNanos)); - } - break; - case EVENT_DRIVEN: - default: - return; - } - - this.schedulingPeriod.set(schedulingPeriod); - } finally { - writeLock.unlock(); - } - } - - @Override - public long getRunDuration(final TimeUnit timeUnit) { - readLock.lock(); - try { - return timeUnit.convert(this.runNanos, TimeUnit.NANOSECONDS); - } finally { - readLock.unlock(); - } - } - - @Override - public void setRunDuration(final long duration, final TimeUnit timeUnit) { - writeLock.lock(); - try { - if (duration < 0) { - throw new IllegalArgumentException("Run Duration must be non-negative value; cannot set to " + timeUnit.toSeconds(duration) + " seconds"); - } - - this.runNanos = timeUnit.toNanos(duration); - } finally { - writeLock.unlock(); - } - } - - @Override - public long getYieldPeriod(final TimeUnit timeUnit) { - return FormatUtils.getTimeDuration(getYieldPeriod(), timeUnit == null ? DEFAULT_TIME_UNIT : timeUnit); - } - - @Override - public String getYieldPeriod() { - return yieldPeriod.get(); - } - - @Override - public void setYieldPeriod(final String yieldPeriod) { - writeLock.lock(); - try { - if (isRunning()) { - throw new IllegalStateException("Cannot modify Processor configuration while the Processor is running"); - } - final long yieldMillis = FormatUtils.getTimeDuration(requireNonNull(yieldPeriod), TimeUnit.MILLISECONDS); - if (yieldMillis < 0) { - throw new IllegalArgumentException("Yield duration must be positive"); - } - this.yieldPeriod.set(yieldPeriod); - } finally { - writeLock.unlock(); - } - } - - /** - * Causes the processor not to be scheduled for some period of time. This duration can be obtained and set via the {@link #getYieldPeriod(TimeUnit)} and {@link #setYieldPeriod(long, TimeUnit)} - * methods. - */ - @Override - public void yield() { - final long yieldMillis = getYieldPeriod(TimeUnit.MILLISECONDS); - yield(yieldMillis, TimeUnit.MILLISECONDS); - - final String yieldDuration = (yieldMillis > 1000) ? (yieldMillis / 1000) + " seconds" : yieldMillis + " milliseconds"; - LoggerFactory.getLogger(processor.getClass()).debug("{} has chosen to yield its resources; will not be scheduled to run again for {}", processor, yieldDuration); - } - - @Override - public void yield(final long period, final TimeUnit timeUnit) { - final long yieldMillis = TimeUnit.MILLISECONDS.convert(period, timeUnit); - yieldExpiration.set(Math.max(yieldExpiration.get(), System.currentTimeMillis() + yieldMillis)); - - processScheduler.yield(this); - } - - /** - * @return the number of milliseconds since Epoch at which time this processor is to once again be scheduled. - */ - @Override - public long getYieldExpiration() { - return yieldExpiration.get(); - } - - @Override - public long getPenalizationPeriod(final TimeUnit timeUnit) { - return FormatUtils.getTimeDuration(getPenalizationPeriod(), timeUnit == null ? DEFAULT_TIME_UNIT : timeUnit); - } - - @Override - public String getPenalizationPeriod() { - return penalizationPeriod.get(); - } - - @Override - public void setPenalizationPeriod(final String penalizationPeriod) { - writeLock.lock(); - try { - if (isRunning()) { - throw new IllegalStateException("Cannot modify Processor configuration while the Processor is running"); - } - final long penalizationMillis = FormatUtils.getTimeDuration(requireNonNull(penalizationPeriod), TimeUnit.MILLISECONDS); - if (penalizationMillis < 0) { - throw new IllegalArgumentException("Penalization duration must be positive"); - } - this.penalizationPeriod.set(penalizationPeriod); - } finally { - writeLock.unlock(); - } - } - - /** - * Determines the number of concurrent tasks that may be running for this processor. - * - * @param taskCount a number of concurrent tasks this processor may have running - * @throws IllegalArgumentException if the given value is less than 1 - */ - @Override - public void setMaxConcurrentTasks(final int taskCount) { - writeLock.lock(); - try { - if (isRunning()) { - throw new IllegalStateException("Cannot modify Processor configuration while the Processor is running"); - } - if (taskCount < 1 && getSchedulingStrategy() != SchedulingStrategy.EVENT_DRIVEN) { - throw new IllegalArgumentException(); - } - if (!triggeredSerially) { - concurrentTaskCount.set(taskCount); - } - } finally { - writeLock.unlock(); - } - } - - @Override - public boolean isTriggeredSerially() { - return triggeredSerially; - } - - /** - * @return the number of tasks that may execute concurrently for this processor - */ - @Override - public int getMaxConcurrentTasks() { - return concurrentTaskCount.get(); - } - - @Override - public LogLevel getBulletinLevel() { - return LogRepositoryFactory.getRepository(getIdentifier()).getObservationLevel(BULLETIN_OBSERVER_ID); - } - - @Override - public void setBulletinLevel(final LogLevel level) { - LogRepositoryFactory.getRepository(getIdentifier()).setObservationLevel(BULLETIN_OBSERVER_ID, level); - } - - @Override - public Set<Connection> getConnections() { - final Set<Connection> allConnections = new HashSet<>(); - readLock.lock(); - try { - for (final Set<Connection> connectionSet : connections.values()) { - allConnections.addAll(connectionSet); - } - } finally { - readLock.unlock(); - } - - return allConnections; - } - - @Override - public List<Connection> getIncomingConnections() { - return incomingConnectionsRef.get(); - } - - @Override - public Set<Connection> getConnections(final Relationship relationship) { - final Set<Connection> applicableConnections; - readLock.lock(); - try { - applicableConnections = connections.get(relationship); - } finally { - readLock.unlock(); - } - return (applicableConnections == null) ? Collections.<Connection>emptySet() : Collections.unmodifiableSet(applicableConnections); - } - - @Override - public void addConnection(final Connection connection) { - Objects.requireNonNull(connection, "connection cannot be null"); - - if (!connection.getSource().equals(this) && !connection.getDestination().equals(this)) { - throw new IllegalStateException("Cannot a connection to a ProcessorNode for which the ProcessorNode is neither the Source nor the Destination"); - } - - writeLock.lock(); - try { - List<Connection> updatedIncoming = null; - if (connection.getDestination().equals(this)) { - // don't add the connection twice. This may occur if we have a self-loop because we will be told - // to add the connection once because we are the source and again because we are the destination. - final List<Connection> incomingConnections = incomingConnectionsRef.get(); - updatedIncoming = new ArrayList<>(incomingConnections); - if (!updatedIncoming.contains(connection)) { - updatedIncoming.add(connection); - } - } - - if (connection.getSource().equals(this)) { - // don't add the connection twice. This may occur if we have a self-loop because we will be told - // to add the connection once because we are the source and again because we are the destination. - if (!destinations.containsKey(connection)) { - for (final Relationship relationship : connection.getRelationships()) { - final Relationship rel = getRelationship(relationship.getName()); - Set<Connection> set = connections.get(rel); - if (set == null) { - set = new HashSet<>(); - connections.put(rel, set); - } - - set.add(connection); - - destinations.put(connection, connection.getDestination()); - } - - final Set<Relationship> autoTerminated = this.undefinedRelationshipsToTerminate.get(); - if (autoTerminated != null) { - autoTerminated.removeAll(connection.getRelationships()); - this.undefinedRelationshipsToTerminate.set(autoTerminated); - } - } - } - - if (updatedIncoming != null) { - incomingConnectionsRef.set(Collections.unmodifiableList(updatedIncoming)); - } - } finally { - writeLock.unlock(); - } - } - - @Override - public boolean hasIncomingConnection() { - return !incomingConnectionsRef.get().isEmpty(); - } - - @Override - public void updateConnection(final Connection connection) throws IllegalStateException { - if (requireNonNull(connection).getSource().equals(this)) { - writeLock.lock(); - try { - // - // update any relationships - // - // first check if any relations were removed. - final List<Relationship> existingRelationships = new ArrayList<>(); - for (final Map.Entry<Relationship, Set<Connection>> entry : connections.entrySet()) { - if (entry.getValue().contains(connection)) { - existingRelationships.add(entry.getKey()); - } - } - - for (final Relationship rel : connection.getRelationships()) { - if (!existingRelationships.contains(rel)) { - // relationship was removed. Check if this is legal. - final Set<Connection> connectionsForRelationship = getConnections(rel); - if (connectionsForRelationship != null && connectionsForRelationship.size() == 1 && this.isRunning() && !isAutoTerminated(rel) && getRelationships().contains(rel)) { - // if we are running and we do not terminate undefined relationships and this is the only - // connection that defines the given relationship, and that relationship is required, - // then it is not legal to remove this relationship from this connection. - throw new IllegalStateException("Cannot remove relationship " + rel.getName() + " from Connection because doing so would invalidate Processor " - + this + ", which is currently running"); - } - } - } - - // remove the connection from any list that currently contains - for (final Set<Connection> list : connections.values()) { - list.remove(connection); - } - - // add the connection in for all relationships listed. - for (final Relationship rel : connection.getRelationships()) { - Set<Connection> set = connections.get(rel); - if (set == null) { - set = new HashSet<>(); - connections.put(rel, set); - } - set.add(connection); - } - - // update to the new destination - destinations.put(connection, connection.getDestination()); - - final Set<Relationship> autoTerminated = this.undefinedRelationshipsToTerminate.get(); - if (autoTerminated != null) { - autoTerminated.removeAll(connection.getRelationships()); - this.undefinedRelationshipsToTerminate.set(autoTerminated); - } - } finally { - writeLock.unlock(); - } - } - - if (connection.getDestination().equals(this)) { - writeLock.lock(); - try { - // update our incoming connections -- we can just remove & re-add the connection to - // update the list. - final List<Connection> incomingConnections = incomingConnectionsRef.get(); - final List<Connection> updatedIncoming = new ArrayList<>(incomingConnections); - updatedIncoming.remove(connection); - updatedIncoming.add(connection); - incomingConnectionsRef.set(Collections.unmodifiableList(updatedIncoming)); - } finally { - writeLock.unlock(); - } - } - } - - @Override - public void removeConnection(final Connection connection) { - boolean connectionRemoved = false; - - if (requireNonNull(connection).getSource().equals(this)) { - for (final Relationship relationship : connection.getRelationships()) { - final Set<Connection> connectionsForRelationship = getConnections(relationship); - if ((connectionsForRelationship == null || connectionsForRelationship.size() <= 1) && isRunning()) { - throw new IllegalStateException("This connection cannot be removed because its source is running and removing it will invalidate this processor"); - } - } - - writeLock.lock(); - try { - for (final Set<Connection> connectionList : this.connections.values()) { - connectionList.remove(connection); - } - - connectionRemoved = (destinations.remove(connection) != null); - } finally { - writeLock.unlock(); - } - } - - if (connection.getDestination().equals(this)) { - writeLock.lock(); - try { - final List<Connection> incomingConnections = incomingConnectionsRef.get(); - if (incomingConnections.contains(connection)) { - final List<Connection> updatedIncoming = new ArrayList<>(incomingConnections); - updatedIncoming.remove(connection); - incomingConnectionsRef.set(Collections.unmodifiableList(updatedIncoming)); - return; - } - } finally { - writeLock.unlock(); - } - } - - if (!connectionRemoved) { - throw new IllegalArgumentException("Cannot remove a connection from a ProcessorNode for which the ProcessorNode is not the Source"); - } - } - - /** - * @param relationshipName name - * @return the relationship for this nodes processor for the given name or creates a new relationship for the given name - */ - @Override - public Relationship getRelationship(final String relationshipName) { - final Relationship specRel = new Relationship.Builder().name(relationshipName).build(); - Relationship returnRel = specRel; - - final Set<Relationship> relationships; - try (final NarCloseable narCloseable = NarCloseable.withNarLoader()) { - relationships = processor.getRelationships(); - } - - for (final Relationship rel : relationships) { - if (rel.equals(specRel)) { - returnRel = rel; - break; - } - } - return returnRel; - } - - @Override - public Processor getProcessor() { - return this.processor; - } - - /** - * @return the Set of destination processors for all relationships excluding any destinations that are this processor itself (self-loops) - */ - public Set<Connectable> getDestinations() { - final Set<Connectable> nonSelfDestinations = new HashSet<>(); - readLock.lock(); - try { - for (final Connectable connectable : destinations.values()) { - if (connectable != this) { - nonSelfDestinations.add(connectable); - } - } - } finally { - readLock.unlock(); - } - return nonSelfDestinations; - } - - public Set<Connectable> getDestinations(final Relationship relationship) { - readLock.lock(); - try { - final Set<Connectable> destinationSet = new HashSet<>(); - final Set<Connection> relationshipConnections = connections.get(relationship); - if (relationshipConnections != null) { - for (final Connection connection : relationshipConnections) { - destinationSet.add(destinations.get(connection)); - } - } - return destinationSet; - } finally { - readLock.unlock(); - } - } - - public Set<Relationship> getUndefinedRelationships() { - final Set<Relationship> undefined = new HashSet<>(); - readLock.lock(); - try { - final Set<Relationship> relationships; - try (final NarCloseable narCloseable = NarCloseable.withNarLoader()) { - relationships = processor.getRelationships(); - } - - if (relationships == null) { - return undefined; - } - for (final Relationship relation : relationships) { - final Set<Connection> connectionSet = this.connections.get(relation); - if (connectionSet == null || connectionSet.isEmpty()) { - undefined.add(relation); - } - } - } finally { - readLock.unlock(); - } - return undefined; - } - - /** - * Determines if the given node is a destination for this node - * - * @param node node - * @return true if is a direct destination node; false otherwise - */ - boolean isRelated(final ProcessorNode node) { - readLock.lock(); - try { - return this.destinations.containsValue(node); - } finally { - readLock.unlock(); - } - } - - @Override - public boolean isRunning() { - readLock.lock(); - try { - return getScheduledState().equals(ScheduledState.RUNNING) || processScheduler.getActiveThreadCount(this) > 0; - } finally { - readLock.unlock(); - } - } - - @Override - public int getActiveThreadCount() { - readLock.lock(); - try { - return processScheduler.getActiveThreadCount(this); - } finally { - readLock.unlock(); - } - } - - @Override - public boolean isValid() { - readLock.lock(); - try { - final ValidationContext validationContext = validationContextFactory.newValidationContext(getProperties(), getAnnotationData()); - - final Collection<ValidationResult> validationResults; - try (final NarCloseable narCloseable = NarCloseable.withNarLoader()) { - validationResults = getProcessor().validate(validationContext); - } - - for (final ValidationResult result : validationResults) { - if (!result.isValid()) { - return false; - } - } - - for (final Relationship undef : getUndefinedRelationships()) { - if (!isAutoTerminated(undef)) { - return false; - } - } - } catch (final Throwable t) { - return false; - } finally { - readLock.unlock(); - } - - return true; - } - - @Override - public Collection<ValidationResult> getValidationErrors() { - final List<ValidationResult> results = new ArrayList<>(); - readLock.lock(); - try { - final ValidationContext validationContext = validationContextFactory.newValidationContext(getProperties(), getAnnotationData()); - - final Collection<ValidationResult> validationResults; - try (final NarCloseable narCloseable = NarCloseable.withNarLoader()) { - validationResults = getProcessor().validate(validationContext); - } - - for (final ValidationResult result : validationResults) { - if (!result.isValid()) { - results.add(result); - } - } - - for (final Relationship relationship : getUndefinedRelationships()) { - if (!isAutoTerminated(relationship)) { - final ValidationResult error = new ValidationResult.Builder() - .explanation("Relationship '" + relationship.getName() + "' is not connected to any component and is not auto-terminated") - .subject("Relationship " + relationship.getName()) - .valid(false) - .build(); - results.add(error); - } - } - } catch (final Throwable t) { - results.add(new ValidationResult.Builder().explanation("Failed to run validation due to " + t.toString()).valid(false).build()); - } finally { - readLock.unlock(); - } - return results; - } - - /** - * Establishes node equality (based on the processor's identifier) - * - * @param other node - * @return true if equal - */ - @Override - public boolean equals(final Object other) { - if (!(other instanceof ProcessorNode)) { - return false; - } - final ProcessorNode on = (ProcessorNode) other; - return new EqualsBuilder().append(identifier.get(), on.getIdentifier()).isEquals(); - } - - @Override - public int hashCode() { - return new HashCodeBuilder(7, 67).append(identifier).toHashCode(); - } - - @Override - public Collection<Relationship> getRelationships() { - try (final NarCloseable narCloseable = NarCloseable.withNarLoader()) { - return getProcessor().getRelationships(); - } - } - - @Override - public String toString() { - try (final NarCloseable narCloseable = NarCloseable.withNarLoader()) { - return getProcessor().toString(); - } - } - - @Override - public ProcessGroup getProcessGroup() { - return processGroup.get(); - } - - @Override - public void setProcessGroup(final ProcessGroup group) { - writeLock.lock(); - try { - this.processGroup.set(group); - } finally { - writeLock.unlock(); - } - } - - @Override - public void onTrigger(final ProcessContext context, final ProcessSessionFactory sessionFactory) { - try (final NarCloseable narCloseable = NarCloseable.withNarLoader()) { - processor.onTrigger(context, sessionFactory); - } - } - - @Override - public ConnectableType getConnectableType() { - return ConnectableType.PROCESSOR; - } - - @Override - public void setScheduledState(final ScheduledState scheduledState) { - this.scheduledState.set(scheduledState); - if (!scheduledState.equals(ScheduledState.RUNNING)) { // if user stops processor, clear yield expiration - yieldExpiration.set(0L); - } - } - - @Override - public void setAnnotationData(final String data) { - writeLock.lock(); - try { - if (isRunning()) { - throw new IllegalStateException("Cannot set AnnotationData while processor is running"); - } - - this.annotationData.set(data); - } finally { - writeLock.unlock(); - } - } - - @Override - public String getAnnotationData() { - return annotationData.get(); - } - - @Override - public Collection<ValidationResult> validate(final ValidationContext validationContext) { - return processor.validate(validationContext); - } - - @Override - public void verifyCanDelete() throws IllegalStateException { - verifyCanDelete(false); - } - - @Override - public void verifyCanDelete(final boolean ignoreConnections) { - readLock.lock(); - try { - if (isRunning()) { - throw new IllegalStateException(this + " is running"); - } - - if (!ignoreConnections) { - for (final Set<Connection> connectionSet : connections.values()) { - for (final Connection connection : connectionSet) { - connection.verifyCanDelete(); - } - } - - for (final Connection connection : incomingConnectionsRef.get()) { - if (connection.getSource().equals(this)) { - connection.verifyCanDelete(); - } else { - throw new IllegalStateException(this + " is the destination of another component"); - } - } - } - } finally { - readLock.unlock(); - } - } - - @Override - public void verifyCanStart() { - readLock.lock(); - try { - switch (getScheduledState()) { - case DISABLED: - throw new IllegalStateException(this + " cannot be started because it is disabled"); - case RUNNING: - throw new IllegalStateException(this + " cannot be started because it is already running"); - case STOPPED: - break; - } - verifyNoActiveThreads(); - - if (!isValid()) { - throw new IllegalStateException(this + " is not in a valid state"); - } - } finally { - readLock.unlock(); - } - } - - @Override - public void verifyCanStart(final Set<ControllerServiceNode> ignoredReferences) { - switch (getScheduledState()) { - case DISABLED: - throw new IllegalStateException(this + " cannot be started because it is disabled"); - case RUNNING: - throw new IllegalStateException(this + " cannot be started because it is already running"); - case STOPPED: - break; - } - verifyNoActiveThreads(); - - final Set<String> ids = new HashSet<>(); - for (final ControllerServiceNode node : ignoredReferences) { - ids.add(node.getIdentifier()); - } - - final Collection<ValidationResult> validationResults = getValidationErrors(ids); - for (final ValidationResult result : validationResults) { - if (!result.isValid()) { - throw new IllegalStateException(this + " cannot be started because it is not valid: " + result); - } - } - } - - @Override - public void verifyCanStop() { - if (getScheduledState() != ScheduledState.RUNNING) { - throw new IllegalStateException(this + " is not scheduled to run"); - } - } - - @Override - public void verifyCanUpdate() { - readLock.lock(); - try { - if (isRunning()) { - throw new IllegalStateException(this + " is not stopped"); - } - } finally { - readLock.unlock(); - } - } - - @Override - public void verifyCanEnable() { - readLock.lock(); - try { - if (getScheduledState() != ScheduledState.DISABLED) { - throw new IllegalStateException(this + " is not disabled"); - } - - verifyNoActiveThreads(); - } finally { - readLock.unlock(); - } - } - - @Override - public void verifyCanDisable() { - readLock.lock(); - try { - if (getScheduledState() != ScheduledState.STOPPED) { - throw new IllegalStateException(this + " is not stopped"); - } - verifyNoActiveThreads(); - } finally { - readLock.unlock(); - } - } - - private void verifyNoActiveThreads() throws IllegalStateException { - final int threadCount = processScheduler.getActiveThreadCount(this); - if (threadCount > 0) { - throw new IllegalStateException(this + " has " + threadCount + " threads still active"); - } - } - - @Override - public void verifyModifiable() throws IllegalStateException { - if (isRunning()) { - throw new IllegalStateException("Cannot modify Processor configuration while the Processor is running"); - } - } + public static final String BULLETIN_OBSERVER_ID = "bulletin-observer"; + + public static final TimeUnit DEFAULT_TIME_UNIT = TimeUnit.MILLISECONDS; + public static final String DEFAULT_YIELD_PERIOD = "1 sec"; + public static final String DEFAULT_PENALIZATION_PERIOD = "30 sec"; + private final AtomicReference<ProcessGroup> processGroup; + private final Processor processor; + private final AtomicReference<String> identifier; + private final Map<Connection, Connectable> destinations; + private final Map<Relationship, Set<Connection>> connections; + private final AtomicReference<Set<Relationship>> undefinedRelationshipsToTerminate; + private final AtomicReference<List<Connection>> incomingConnectionsRef; + private final ReentrantReadWriteLock rwLock; + private final Lock readLock; + private final Lock writeLock; + private final AtomicBoolean isolated; + private final AtomicBoolean lossTolerant; + private final AtomicReference<ScheduledState> scheduledState; + private final AtomicReference<String> comments; + private final AtomicReference<String> name; + private final AtomicReference<Position> position; + private final AtomicReference<String> annotationData; + private final AtomicReference<String> schedulingPeriod; // stored as string so it's presented to user as they entered it + private final AtomicReference<String> yieldPeriod; + private final AtomicReference<String> penalizationPeriod; + private final AtomicReference<Map<String, String>> style; + private final AtomicInteger concurrentTaskCount; + private final AtomicLong yieldExpiration; + private final AtomicLong schedulingNanos; + private final boolean triggerWhenEmpty; + private final boolean sideEffectFree; + private final boolean triggeredSerially; + private final boolean triggerWhenAnyDestinationAvailable; + private final boolean eventDrivenSupported; + private final boolean batchSupported; + private final Requirement inputRequirement; + private final ValidationContextFactory validationContextFactory; + private final ProcessScheduler processScheduler; + private long runNanos = 0L; + + private SchedulingStrategy schedulingStrategy; // guarded by read/write lock + + @SuppressWarnings("deprecation") + public StandardProcessorNode(final Processor processor, final String uuid, final ValidationContextFactory validationContextFactory, + final ProcessScheduler scheduler, final ControllerServiceProvider controllerServiceProvider) { + super(processor, uuid, validationContextFactory, controllerServiceProvider); + + this.processor = processor; + identifier = new AtomicReference<>(uuid); + destinations = new HashMap<>(); + connections = new HashMap<>(); + incomingConnectionsRef = new AtomicReference<List<Connection>>(new ArrayList<Connection>()); + scheduledState = new AtomicReference<>(ScheduledState.STOPPED); + rwLock = new ReentrantReadWriteLock(false); + readLock = rwLock.readLock(); + writeLock = rwLock.writeLock(); + lossTolerant = new AtomicBoolean(false); + final Set<Relationship> emptySetOfRelationships = new HashSet<>(); + undefinedRelationshipsToTerminate = new AtomicReference<>(emptySetOfRelationships); + comments = new AtomicReference<>(""); + name = new AtomicReference<>(processor.getClass().getSimpleName()); + schedulingPeriod = new AtomicReference<>("0 sec"); + schedulingNanos = new AtomicLong(MINIMUM_SCHEDULING_NANOS); + yieldPeriod = new AtomicReference<>(DEFAULT_YIELD_PERIOD); + yieldExpiration = new AtomicLong(0L); + concurrentTaskCount = new AtomicInteger(1); + position = new AtomicReference<>(new Position(0D, 0D)); + style = new AtomicReference<>(Collections.unmodifiableMap(new HashMap<String, String>())); + this.processGroup = new AtomicReference<>(); + processScheduler = scheduler; + annotationData = new AtomicReference<>(); + isolated = new AtomicBoolean(false); + penalizationPeriod = new AtomicReference<>(DEFAULT_PENALIZATION_PERIOD); + + final Class<?> procClass = processor.getClass(); + triggerWhenEmpty = procClass.isAnnotationPresent(TriggerWhenEmpty.class) || procClass.isAnnotationPresent(org.apache.nifi.processor.annotation.TriggerWhenEmpty.class); + sideEffectFree = procClass.isAnnotationPresent(SideEffectFree.class) || procClass.isAnnotationPresent(org.apache.nifi.processor.annotation.SideEffectFree.class); + batchSupported = procClass.isAnnotationPresent(SupportsBatching.class) || procClass.isAnnotationPresent(org.apache.nifi.processor.annotation.SupportsBatching.class); + triggeredSerially = procClass.isAnnotationPresent(TriggerSerially.class) || procClass.isAnnotationPresent(org.apache.nifi.processor.annotation.TriggerSerially.class); + triggerWhenAnyDestinationAvailable = procClass.isAnnotationPresent(TriggerWhenAnyDestinationAvailable.class) + || procClass.isAnnotationPresent(org.apache.nifi.processor.annotation.TriggerWhenAnyDestinationAvailable.class); + this.validationContextFactory = validationContextFactory; + eventDrivenSupported = (procClass.isAnnotationPresent(EventDriven.class) + || procClass.isAnnotationPresent(org.apache.nifi.processor.annotation.EventDriven.class)) && !triggeredSerially && !triggerWhenEmpty; + + final boolean inputRequirementPresent = procClass.isAnnotationPresent(InputRequirement.class); + if (inputRequirementPresent) { + inputRequirement = procClass.getAnnotation(InputRequirement.class).value(); + } else { + inputRequirement = Requirement.INPUT_ALLOWED; + } + + schedulingStrategy = SchedulingStrategy.TIMER_DRIVEN; + } + + /** + * @return comments about this specific processor instance + */ + @Override + public String getComments() { + return comments.get(); + } + + /** + * Provides and opportunity to retain information about this particular processor instance + * + * @param comments new comments + */ + @Override + public void setComments(final String comments) { + writeLock.lock(); + try { + if (isRunning()) { + throw new IllegalStateException("Cannot modify Processor configuration while the Processor is running"); + } + this.comments.set(comments); + } finally { + writeLock.unlock(); + } + } + + @Override + public ScheduledState getScheduledState() { + return scheduledState.get(); + } + + @Override + public Position getPosition() { + return position.get(); + } + + @Override + public void setPosition(Position position) { + this.position.set(position); + } + + @Override + public Map<String, String> getStyle() { + return style.get(); + } + + @Override + public void setStyle(final Map<String, String> style) { + if (style != null) { + this.style.set(Collections.unmodifiableMap(new HashMap<>(style))); + } + } + + @Override + public String getIdentifier() { + return identifier.get(); + } + + /** + * @return if true flow file content generated by this processor is considered loss tolerant + */ + @Override + public boolean isLossTolerant() { + return lossTolerant.get(); + } + + @Override + public boolean isIsolated() { + return isolated.get(); + } + + /** + * @return true if the processor has the {@link TriggerWhenEmpty} annotation, false otherwise. + */ + @Override + public boolean isTriggerWhenEmpty() { + return triggerWhenEmpty; + } + + /** + * @return true if the processor has the {@link SideEffectFree} annotation, false otherwise. + */ + @Override + public boolean isSideEffectFree() { + return sideEffectFree; + } + + @Override + public boolean isHighThroughputSupported() { + return batchSupported; + } + + /** + * @return true if the processor has the {@link TriggerWhenAnyDestinationAvailable} annotation, false otherwise. + */ + @Override + public boolean isTriggerWhenAnyDestinationAvailable() { + return triggerWhenAnyDestinationAvailable; + } + + /** + * Indicates whether flow file content made by this processor must be persisted + * + * @param lossTolerant tolerant + */ + @Override + public void setLossTolerant(final boolean lossTolerant) { + writeLock.lock(); + try { + if (isRunning()) { + throw new IllegalStateException("Cannot modify Processor configuration while the Processor is running"); + } + this.lossTolerant.set(lossTolerant); + } finally { + writeLock.unlock(); + } + } + + /** + * Indicates whether the processor runs on only the primary node. + * + * @param isolated isolated + */ + public void setIsolated(final boolean isolated) { + writeLock.lock(); + try { + if (isRunning()) { + throw new IllegalStateException("Cannot modify Processor configuration while the Processor is running"); + } + this.isolated.set(isolated); + } finally { + writeLock.unlock(); + } + } + + @Override + public boolean isAutoTerminated(final Relationship relationship) { + final Set<Relationship> terminatable = undefinedRelationshipsToTerminate.get(); + if (terminatable == null) { + return false; + } + return terminatable.contains(relationship); + } + + @Override + public void setAutoTerminatedRelationships(final Set<Relationship> terminate) { + writeLock.lock(); + try { + if (isRunning()) { + throw new IllegalStateException("Cannot modify Processor configuration while the Processor is running"); + } + + for (final Relationship rel : terminate) { + if (!getConnections(rel).isEmpty()) { + throw new IllegalStateException("Cannot mark relationship '" + rel.getName() + "' as auto-terminated because Connection already exists with this relationship"); + } + } + undefinedRelationshipsToTerminate.set(new HashSet<>(terminate)); + } finally { + writeLock.unlock(); + } + } + + /** + * @return an unmodifiable Set that contains all of the ProcessorRelationship objects that are configured to be auto-terminated + */ + @Override + public Set<Relationship> getAutoTerminatedRelationships() { + Set<Relationship> relationships = undefinedRelationshipsToTerminate.get(); + if (relationships == null) { + relationships = new HashSet<>(); + } + return Collections.unmodifiableSet(relationships); + } + + @Override + public String getName() { + return name.get(); + } + + /** + * @return the value of the processor's {@link CapabilityDescription} annotation, if one exists, else <code>null</code>. + */ + @SuppressWarnings("deprecation") + public String getProcessorDescription() { + CapabilityDescription capDesc = processor.getClass().getAnnotation(CapabilityDescription.class); + String description = null; + if (capDesc != null) { + description = capDesc.value(); + } else { + final org.apache.nifi.processor.annotation.CapabilityDescription deprecatedCapDesc + = processor.getClass().getAnnotation(org.apache.nifi.processor.annotation.CapabilityDescription.class); + if (deprecatedCapDesc != null) { + description = deprecatedCapDesc.value(); + } + } + + return description; + } + + @Override + public void setName(final String name) { + writeLock.lock(); + try { + if (isRunning()) { + throw new IllegalStateException("Cannot modify Processor configuration while the Processor is running"); + } + this.name.set(name); + } finally { + writeLock.unlock(); + } + } + + /** + * @param timeUnit determines the unit of time to represent the scheduling period. If null will be reported in units of {@link #DEFAULT_SCHEDULING_TIME_UNIT} + * @return the schedule period that should elapse before subsequent cycles of this processor's tasks + */ + @Override + public long getSchedulingPeriod(final TimeUnit timeUnit) { + return timeUnit.convert(schedulingNanos.get(), TimeUnit.NANOSECONDS); + } + + @Override + public boolean isEventDrivenSupported() { + readLock.lock(); + try { + return this.eventDrivenSupported; + } finally { + readLock.unlock(); + } + } + + /** + * Updates the Scheduling Strategy used for this Processor + * + * @param schedulingStrategy strategy + * + * @throws IllegalArgumentException if the SchedulingStrategy is not not applicable for this Processor + */ + @Override + public void setSchedulingStrategy(final SchedulingStrategy schedulingStrategy) { + writeLock.lock(); + try { + if (schedulingStrategy == SchedulingStrategy.EVENT_DRIVEN && !eventDrivenSupported) { + // not valid. Just ignore it. We don't throw an Exception because if a developer changes a Processor so that + // it no longer supports EventDriven mode, we don't want the app to fail to startup if it was already in Event-Driven + // Mode. Instead, we will simply leave it in Timer-Driven mode + return; + } + + this.schedulingStrategy = schedulingStrategy; + setIsolated(schedulingStrategy == SchedulingStrategy.PRIMARY_NODE_ONLY); + } finally { + writeLock.unlock(); + } + } + + /** + * @return the currently configured scheduling strategy + */ + @Override + public SchedulingStrategy getSchedulingStrategy() { + readLock.lock(); + try { + return this.schedulingStrategy; + } finally { + readLock.unlock(); + } + } + + @Override + public String getSchedulingPeriod() { + return schedulingPeriod.get(); + } + + @Override + public void setScheduldingPeriod(final String schedulingPeriod) { + writeLock.lock(); + try { + if (isRunning()) { + throw new IllegalStateException("Cannot modify Processor configuration while the Processor is running"); + } + + switch (schedulingStrategy) { + case CRON_DRIVEN: { + try { + new CronExpression(schedulingPeriod); + } catch (final Exception e) { + throw new IllegalArgumentException("Scheduling Period is not a valid cron expression: " + schedulingPeriod); + } + } + break; + case PRIMARY_NODE_ONLY: + case TIMER_DRIVEN: { + final long schedulingNanos = FormatUtils.getTimeDuration(requireNonNull(schedulingPeriod), TimeUnit.NANOSECONDS); + if (schedulingNanos < 0) { + throw new IllegalArgumentException("Scheduling Period must be positive"); + } + this.schedulingNanos.set(Math.max(MINIMUM_SCHEDULING_NANOS, schedulingNanos)); + } + break; + case EVENT_DRIVEN: + default: + return; + } + + this.schedulingPeriod.set(schedulingPeriod); + } finally { + writeLock.unlock(); + } + } + + @Override + public long getRunDuration(final TimeUnit timeUnit) { + readLock.lock(); + try { + return timeUnit.convert(this.runNanos, TimeUnit.NANOSECONDS); + } finally { + readLock.unlock(); + } + } + + @Override + public void setRunDuration(final long duration, final TimeUnit timeUnit) { + writeLock.lock(); + try { + if (duration < 0) { + throw new IllegalArgumentException("Run Duration must be non-negative value; cannot set to " + timeUnit.toSeconds(duration) + " seconds"); + } + + this.runNanos = timeUnit.toNanos(duration); + } finally { + writeLock.unlock(); + } + } + + @Override + public long getYieldPeriod(final TimeUnit timeUnit) { + return FormatUtils.getTimeDuration(getYieldPeriod(), timeUnit == null ? DEFAULT_TIME_UNIT : timeUnit); + } + + @Override + public String getYieldPeriod() { + return yieldPeriod.get(); + } + + @Override + public void setYieldPeriod(final String yieldPeriod) { + writeLock.lock(); + try { + if (isRunning()) { + throw new IllegalStateException("Cannot modify Processor configuration while the Processor is running"); + } + final long yieldMillis = FormatUtils.getTimeDuration(requireNonNull(yieldPeriod), TimeUnit.MILLISECONDS); + if (yieldMillis < 0) { + throw new IllegalArgumentException("Yield duration must be positive"); + } + this.yieldPeriod.set(yieldPeriod); + } finally { + writeLock.unlock(); + } + } + + /** + * Causes the processor not to be scheduled for some period of time. This duration can be obtained and set via the {@link #getYieldPeriod(TimeUnit)} and {@link #setYieldPeriod(long, TimeUnit)} + * methods. + */ + @Override + public void yield() { + final long yieldMillis = getYieldPeriod(TimeUnit.MILLISECONDS); + yield(yieldMillis, TimeUnit.MILLISECONDS); + + final String yieldDuration = (yieldMillis > 1000) ? (yieldMillis / 1000) + " seconds" : yieldMillis + " milliseconds"; + LoggerFactory.getLogger(processor.getClass()).debug("{} has chosen to yield its resources; will not be scheduled to run again for {}", processor, yieldDuration); + } + + @Override + public void yield(final long period, final TimeUnit timeUnit) { + final long yieldMillis = TimeUnit.MILLISECONDS.convert(period, timeUnit); + yieldExpiration.set(Math.max(yieldExpiration.get(), System.currentTimeMillis() + yieldMillis)); + + processScheduler.yield(this); + } + + /** + * @return the number of milliseconds since Epoch at which time this processor is to once again be scheduled. + */ + @Override + public long getYieldExpiration() { + return yieldExpiration.get(); + } + + @Override + public long getPenalizationPeriod(final TimeUnit timeUnit) { + return FormatUtils.getTimeDuration(getPenalizationPeriod(), timeUnit == null ? DEFAULT_TIME_UNIT : timeUnit); + } + + @Override + public String getPenalizationPeriod() { + return penalizationPeriod.get(); + } + + @Override + public void setPenalizationPeriod(final String penalizationPeriod) { + writeLock.lock(); + try { + if (isRunning()) { + throw new IllegalStateException("Cannot modify Processor configuration while the Processor is running"); + } + final long penalizationMillis = FormatUtils.getTimeDuration(requireNonNull(penalizationPeriod), TimeUnit.MILLISECONDS); + if (penalizationMillis < 0) { + throw new IllegalArgumentException("Penalization duration must be positive"); + } + this.penalizationPeriod.set(penalizationPeriod); + } finally { + writeLock.unlock(); + } + } + + /** + * Determines the number of concurrent tasks that may be running for this processor. + * + * @param taskCount a number of concurrent tasks this processor may have running + * @throws IllegalArgumentException if the given value is less than 1 + */ + @Override + public void setMaxConcurrentTasks(final int taskCount) { + writeLock.lock(); + try { + if (isRunning()) { + throw new IllegalStateException("Cannot modify Processor configuration while the Processor is running"); + } + if (taskCount < 1 && getSchedulingStrategy() != SchedulingStrategy.EVENT_DRIVEN) { + throw new IllegalArgumentException(); + } + if (!triggeredSerially) { + concurrentTaskCount.set(taskCount); + } + } finally { + writeLock.unlock(); + } + } + + @Override + public boolean isTriggeredSerially() { + return triggeredSerially; + } + + /** + * @return the number of tasks that may execute concurrently for this processor + */ + @Override + public int getMaxConcurrentTasks() { + return concurrentTaskCount.get(); + } + + @Override + public LogLevel getBulletinLevel() { + return LogRepositoryFactory.getRepository(getIdentifier()).getObservationLevel(BULLETIN_OBSERVER_ID); + } + + @Override + public void setBulletinLevel(final LogLevel level) { + LogRepositoryFactory.getRepository(getIdentifier()).setObservationLevel(BULLETIN_OBSERVER_ID, level); + } + + @Override + public Set<Connection> getConnections() { + final Set<Connection> allConnections = new HashSet<>(); + readLock.lock(); + try { + for (final Set<Connection> connectionSet : connections.values()) { + allConnections.addAll(connectionSet); + } + } finally { + readLock.unlock(); + } + + return allConnections; + } + + @Override + public List<Connection> getIncomingConnections() { + return incomingConnectionsRef.get(); + } + + @Override + public Set<Connection> getConnections(final Relationship relationship) { + final Set<Connection> applicableConnections; + readLock.lock(); + try { + applicableConnections = connections.get(relationship); + } finally { + readLock.unlock(); + } + return (applicableConnections == null) ? Collections.<Connection>emptySet() : Collections.unmodifiableSet(applicableConnections); + } + + @Override + public void addConnection(final Connection connection) { + Objects.requireNonNull(connection, "connection cannot be null"); + + if (!connection.getSource().equals(this) && !connection.getDestination().equals(this)) { + throw new IllegalStateException("Cannot a connection to a ProcessorNode for which the ProcessorNode is neither the Source nor the Destination"); + } + + writeLock.lock(); + try { + List<Connection> updatedIncoming = null; + if (connection.getDestination().equals(this)) { + // don't add the connection twice. This may occur if we have a self-loop because we will be told + // to add the connection once because we are the source and again because we are the destination. + final List<Connection> incomingConnections = incomingConnectionsRef.get(); + updatedIncoming = new ArrayList<>(incomingConnections); + if (!updatedIncoming.contains(connection)) { + updatedIncoming.add(connection); + } + } + + if (connection.getSource().equals(this)) { + // don't add the connection twice. This may occur if we have a self-loop because we will be told + // to add the connection once because we are the source and again because we are the destination. + if (!destinations.containsKey(connection)) { + for (final Relationship relationship : connection.getRelationships()) { + final Relationship rel = getRelationship(relationship.getName()); + Set<Connection> set = connections.get(rel); + if (set == null) { + set = new HashSet<>(); + connections.put(rel, set); + } + + set.add(connection); + + destinations.put(connection, connection.getDestination()); + } + + final Set<Relationship> autoTerminated = this.undefinedRelationshipsToTerminate.get(); + if (autoTerminated != null) { + autoTerminated.removeAll(connection.getRelationships()); + this.undefinedRelationshipsToTerminate.set(autoTerminated); + } + } + } + + if (updatedIncoming != null) { + incomingConnectionsRef.set(Collections.unmodifiableList(updatedIncoming)); + } + } finally { + writeLock.unlock(); + } + } + + @Override + public boolean hasIncomingConnection() { + return !incomingConnectionsRef.get().isEmpty(); + } + + @Override + public void updateConnection(final Connection connection) throws IllegalStateException { + if (requireNonNull(connection).getSource().equals(this)) { + writeLock.lock(); + try { + // + // update any relationships + // + // first check if any relations were removed. + final List<Relationship> existingRelationships = new ArrayList<>(); + for (final Map.Entry<Relationship, Set<Connection>> entry : connections.entrySet()) { + if (entry.getValue().contains(connection)) { + existingRelationships.add(entry.getKey()); + } + } + + for (final Relationship rel : connection.getRelationships()) { + if (!existingRelationships.contains(rel)) { + // relationship was removed. Check if this is legal. + final Set<Connection> connectionsForRelationship = getConnections(rel); + if (connectionsForRelationship != null && connectionsForRelationship.size() == 1 && this.isRunning() && !isAutoTerminated(rel) && getRelationships().contains(rel)) { + // if we are running and we do not terminate undefined relationships and this is the only + // connection that defines the given relationship, and that relationship is required, + // then it is not legal to remove this relationship from this connection. + throw new IllegalStateException("Cannot remove relationship " + rel.getName() + " from Connection because doing so would invalidate Processor " + + this + ", which is currently running"); + } + } + } + + // remove the connection from any list that currently contains + for (final Set<Connection> list : connections.values()) { + list.remove(connection); + } + + // add the connection in for all relationships listed. + for (final Relationship rel : connection.getRelationships()) { + Set<Connection> set = connections.get(rel); + if (set == null) { + set = new HashSet<>(); + connections.put(rel, set); + } + set.add(connection); + } + + // update to the new destination + destinations.put(connection, connection.getDestination()); + + final Set<Relationship> autoTerminated = this.undefinedRelationshipsToTerminate.get(); + if (autoTerminated != null) { + autoTerminated.removeAll(connection.getRelationships()); + this.undefinedRelationshipsToTerminate.set(autoTerminated); + } + } finally { + writeLock.unlock(); + } + } + + if (connection.getDestination().equals(this)) { + writeLock.lock(); + try { + // update our incoming connections -- we can just remove & re-add the connection to + // update the list. + final List<Connection> incomingConnections = incomingConnectionsRef.get(); + final List<Connection> updatedIncoming = new ArrayList<>(incomingConnections); + updatedIncoming.remove(connection); + updatedIncoming.add(connection); + incomingConnectionsRef.set(Collections.unmodifiableList(updatedIncoming)); + } finally { + writeLock.unlock(); + } + } + } + + @Override + public void removeConnection(final Connection connection) { + boolean connectionRemoved = false; + + if (requireNonNull(connection).getSource().equals(this)) { + for (final Relationship relationship : connection.getRelationships()) { + final Set<Connection> connectionsForRelationship = getConnections(relationship); + if ((connectionsForRelationship == null || connectionsForRelationship.size() <= 1) && isRunning()) { + throw new IllegalStateException("This connection cannot be removed because its source is running and removing it will invalidate this processor"); + } + } + + writeLock.lock(); + try { + for (final Set<Connection> connectionList : this.connections.values()) { + connectionList.remove(connection); + } + + connectionRemoved = (destinations.remove(connection) != null); + } finally { + writeLock.unlock(); + } + } + + if (connection.getDestination().equals(this)) { + writeLock.lock(); + try { + final List<Connection> incomingConnections = incomingConnectionsRef.get(); + if (incomingConnections.contains(connection)) { + final List<Connection> updatedIncoming = new ArrayList<>(incomingConnections); + updatedIncoming.remove(connection); + incomingConnectionsRef.set(Collections.unmodifiableList(updatedIncoming)); + return; + } + } finally { + writeLock.unlock(); + } + } + + if (!connectionRemoved) { + throw new IllegalArgumentException("Cannot remove a connection from a ProcessorNode for which the ProcessorNode is not the Source"); + } + } + + /** + * @param relationshipName name + * @return the relationship for this nodes processor for the given name or creates a new relationship for the given name + */ + @Override + public Relationship getRelationship(final String relationshipName) { + final Relationship specRel = new Relationship.Builder().name(relationshipName).build(); + Relationship returnRel = specRel; + + final Set<Relationship> relationships; + try (final NarCloseable narCloseable = NarCloseable.withNarLoader()) { + relationships = processor.getRelationships(); + } + + for (final Relationship rel : relationships) { + if (rel.equals(specRel)) { + returnRel = rel; + break; + } + } + return returnRel; + } + + @Override + public Processor getProcessor() { + return this.processor; + } + + /** + * @return the Set of destination processors for all relationships excluding any destinations that are this processor itself (self-loops) + */ + public Set<Connectable> getDestinations() { + final Set<Connectable> nonSelfDestinations = new HashSet<>(); + readLock.lock(); + try { + for (final Connectable connectable : destinations.values()) { + if (connectable != this) { + nonSelfDestinations.add(connectable); + } + } + } finally { + readLock.unlock(); + } + return nonSelfDestinations; + } + + public Set<Connectable> getDestinations(final Relationship relationship) { + readLock.lock(); + try { + final Set<Connectable> destinationSet = new HashSet<>(); + final Set<Connection> relationshipConnections = connections.get(relationship); + if (relationshipConnections != null) { + for (final Connection connection : relationshipConnections) { + destinationSet.add(destinations.get(connection)); + } + } + return destinationSet; + } finally { + readLock.unlock(); + } + } + + public Set<Relationship> getUndefinedRelationships() { + final Set<Relationship> undefined = new HashSet<>(); + readLock.lock(); + try { + final Set<Relationship> relationships; + try (final NarCloseable narCloseable = NarCloseable.withNarLoader()) { + relationships = processor.getRelationships(); + } + + if (relationships == null) { + return undefined; + } + for (final Relationship relation : relationships) { + final Set<Connection> connectionSet = this.connections.get(relation); + if (connectionSet == null || connectionSet.isEmpty()) { + undefined.add(relation); + } + } + } finally { + readLock.unlock(); + } + return undefined; + } + + /** + * Determines if the given node is a destination for this node + * + * @param node node + * @return true if is a direct destination node; false otherwise + */ + boolean isRelated(final ProcessorNode node) { + readLock.lock(); + try { + return this.destinations.containsValue(node); + } finally { + readLock.unlock(); + } + } + + @Override + public boolean isRunning() { + readLock.lock(); + try { + return getScheduledState().equals(ScheduledState.RUNNING) || processScheduler.getActiveThreadCount(this) > 0; + } finally { + readLock.unlock(); + } + } + + @Override + public int getActiveThreadCount() { + readLock.lock(); + try { + return processScheduler.getActiveThreadCount(this); + } finally { + readLock.unlock(); + } + } + + @Override + public boolean isValid() { + readLock.lock(); + try { + final ValidationContext validationContext = validationContextFactory.newValidationContext(getProperties(), getAnnotationData()); + + final Collection<ValidationResult> validationResults; + try (final NarCloseable narCloseable = NarCloseable.withNarLoader()) { + validationResults = getProcessor().validate(validationContext); + } + + for (final ValidationResult result : validationResults) { + if (!result.isValid()) { + return false; + } + } + + for (final Relationship undef : getUndefinedRelationships()) { + if (!isAutoTerminated(undef)) { + return false; + } + } + + switch (getInputRequirement()) { + case INPUT_ALLOWED: + break; + case INPUT_FORBIDDEN: { + if (!getIncomingConnections().isEmpty()) { + return false; + } + break; + } + case INPUT_REQUIRED: { + if (getIncomingConnections().isEmpty()) { + return false; + } + break; + } + } + } catch (final Throwable t) { + return false; + } finally { + readLock.unlock(); + } + + return true; + } + + @Override + public Collection<ValidationResult> getValidationErrors() { + final List<ValidationResult> results = new ArrayList<>(); + readLock.lock(); + try { + final ValidationContext validationContext = validationContextFactory.newValidationContext(getProperties(), getAnnotationData()); + + final Collection<ValidationResult> validationResults; + try (final NarCloseable narCloseable = NarCloseable.withNarLoader()) { + validationResults = getProcessor().validate(validationContext); + } + + for (final ValidationResult result : validationResults) { + if (!result.isValid()) { + results.add(result); + } + } + + for (final Relationship relationship : getUndefinedRelationships()) { + if (!isAutoTerminated(relationship)) { + final ValidationResult error = new ValidationResult.Builder() + .explanation("Relationship '" + relationship.getName() + "' is not connected to any component and is not auto-terminated") + .subject("Relationship " + relationship.getName()) + .valid(false) + .build(); + results.add(error); + } + } + + switch (getInputRequirement()) { + case INPUT_ALLOWED: + break; + case INPUT_FORBIDDEN: { + final int incomingConnCount = getIncomingConnections().size(); + if (incomingConnCount != 0) { + results.add(new ValidationResult.Builder() + .explanation("Processor does not accept Incoming Connections but is currently configured with " + incomingConnCount + " Incoming Connections") + .subject("Incoming Connections") + .valid(false) + .build()); + } + break; + } + case INPUT_REQUIRED: { + if (getIncomingConnections().isEmpty()) { + results.add(new ValidationResult.Builder() + .explanation("Processor required at least one Incoming Connection in order to perform its function but currently has no Incoming Connection") + .subject("Incoming Connections") + .valid(false) + .build()); + } + break; + } + } + } catch (final Throwable t) { + results.add(new ValidationResult.Builder().explanation("Failed to run validation due to " + t.toString()).valid(false).build()); + } finally { + readLock.unlock(); + } + return results; + } + + @Override + public Requirement getInputRequirement() { + return inputRequirement; + } + + /** + * Establishes node equality (based on the processor's identifier) + * + * @param other node + * @return true if equal + */ + @Override + public boolean equals(final Object other) { + if (!(other instanceof ProcessorNode)) { + return false; + } + final ProcessorNode on = (ProcessorNode) other; + return new EqualsBuilder().append(identifier.get(), on.getIdentifier()).isEquals(); + } + + @Override + public int hashCode() { + return new HashCodeBuilder(7, 67).append(identifier).toHashCode(); + } + + @Override + public Collection<Relationship> getRelationships() { + try (final NarCloseable narCloseable = NarCloseable.withNarLoader()) { + return getProcessor().getRelationships(); + } + } + + @Override + public String toString() { + try (final NarCloseable narCloseable = NarCloseable.withNarLoader()) { + return getProcessor().toString(); + } + } + + @Override + public ProcessGroup getProcessGroup() { + return processGroup.get(); + } + + @Override + public void setProcessGroup(final ProcessGroup group) { + writeLock.lock(); + try { + this.processGroup.set(group); + } finally { + writeLock.unlock(); + } + } + + @Override + public void onTrigger(final ProcessContext context, final ProcessSessionFactory sessionFactory) { + try (final NarCloseable narCloseable = NarCloseable.withNarLoader()) { + processor.onTrigger(context, sessionFactory); + } + } + + @Override + public ConnectableType getConnectableType() { + return ConnectableType.PROCESSOR; + } + + @Override + public void setScheduledState(final ScheduledState scheduledState) { + this.scheduledState.set(scheduledState); + if (!scheduledState.equals(ScheduledState.RUNNING)) { // if user stops processor, clear yield expiration + yieldExpiration.set(0L); + } + } + + @Override + public void setAnnotationData(final String data) { + writeLock.lock(); + try { + if (isRunning()) { + throw new IllegalStateException("Cannot set AnnotationData while processor is running"); + } + + this.annotationData.set(data); + } finally { + writeLock.unlock(); + } + } + + @Override + public String getAnnotationData() { + return annotationData.get(); + } + + @Override + public Collection<ValidationResult> validate(final ValidationContext validationContext) { + return getValidationErrors(); + } + + @Override + public void verifyCanDelete() throws IllegalStateException { + verifyCanDelete(false); + } + + @Override + public void verifyCanDelete(final boolean ignoreConnections) { + readLock.lock(); + try { + if (isRunning()) { + throw new IllegalStateException(this + " is running"); + } + + if (!ignoreConnections) { + for (final Set<Connection> connectionSet : connections.values()) { + for (final Connection connection : connectionSet) { + connection.verifyCanDelete(); + } + } + + for (final Connection connection : incomingConnectionsRef.get()) { + if (connection.getSource().equals(this)) { + connection.verifyCanDelete(); + } else { + throw new IllegalStateException(this + " is the destination of another component"); + } + } + } + } finally { + readLock.unlock(); + } + } + + @Override + public void verifyCanStart() { + readLock.lock(); + try { + switch (getScheduledState()) { + case DISABLED: + throw new IllegalStateException(this + " cannot be started because it is disabled"); + case RUNNING: + throw new IllegalStateException(this + " cannot be started because it is already running"); + case STOPPED: + break; + } + verifyNoActiveThreads(); + + if (!isValid()) { + throw new IllegalStateException(this + " is not in a valid state"); + } + } finally { + readLock.unlock(); + } + } + + @Override + public void verifyCanStart(final Set<ControllerServiceNode> ignoredReferences) { + switch (getScheduledState()) { + case DISABLED: + throw new IllegalStateException(this + " cannot be started because it is disabled"); + case RUNNING: + throw new IllegalStateException(this + " cannot be started because it is already running"); + case STOPPED: + break; + } + verifyNoActiveThreads(); + + final Set<String> ids = new HashSet<>(); + for (final ControllerServiceNode node : ignoredReferences) { + ids.add(node.getIdentifier()); + } + + final Collection<ValidationResult> validationResults = getValidationErrors(ids); + for (final ValidationResult result : validationResults) { + if (!result.isValid()) { + throw new IllegalStateException(this + " cannot be started because it is not valid: " + result); + } + } + } + + @Override + public void verifyCanStop() { + if (getScheduledState() != ScheduledState.RUNNING) { + throw new IllegalStateException(this + " is not scheduled to run"); + } + } + + @Override + public void verifyCanUpdate() { + readLock.lock(); + try { + if (isRunning()) { + throw new IllegalStateException(this + " is not stopped"); + } + } finally { + readLock.unlock(); + } + } + + @Override + public void verifyCanEnable() { + readLock.lock(); + try { + if (getScheduledState() != ScheduledState.DISABLED) { + throw new IllegalStateException(this + " is not disabled"); + } + + verifyNoActiveThreads(); + } finally { + readLock.unlock(); + } + } + + @Override + public void verifyCanDisable() { + readLock.lock(); + try { + if (getScheduledState() != ScheduledState.STOPPED) { + throw new IllegalStateException(this + " is not stopped"); + } + verifyNoActiveThreads(); + } finally { + readLock.unlock(); + } + } + + private void verifyNoActiveThreads() throws IllegalStateException { + final int threadCount = processScheduler.getActiveThreadCount(this); + if (threadCount > 0) { + throw new IllegalStateException(this + " has " + threadCount + " threads still active"); + } + } + + @Override + public void verifyModifiable() throws IllegalStateException { + if (isRunning()) { + throw new IllegalStateException("Cannot modify Processor configuration while the Processor is running"); + } + } }
http://git-wip-us.apache.org/repos/asf/nifi/blob/034ee6de/nifi-nar-bundles/nifi-geo-bundle/nifi-geo-processors/src/main/java/org/apache/nifi/processors/GeoEnrichIP.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-geo-bundle/nifi-geo-processors/src/main/java/org/apache/nifi/processors/GeoEnrichIP.java b/nifi-nar-bundles/nifi-geo-bundle/nifi-geo-processors/src/main/java/org/apache/nifi/processors/GeoEnrichIP.java index b766878..eccff79 100644 --- a/nifi-nar-bundles/nifi-geo-bundle/nifi-geo-processors/src/main/java/org/apache/nifi/processors/GeoEnrichIP.java +++ b/nifi-nar-bundles/nifi-geo-bundle/nifi-geo-processors/src/main/java/org/apache/nifi/processors/GeoEnrichIP.java @@ -31,6 +31,8 @@ import java.util.concurrent.atomic.AtomicReference; import org.apache.commons.lang3.StringUtils; import org.apache.nifi.annotation.behavior.EventDriven; +import org.apache.nifi.annotation.behavior.InputRequirement; +import org.apache.nifi.annotation.behavior.InputRequirement.Requirement; import org.apache.nifi.annotation.behavior.SideEffectFree; import org.apache.nifi.annotation.behavior.SupportsBatching; import org.apache.nifi.annotation.behavior.WritesAttribute; @@ -59,6 +61,7 @@ import com.maxmind.geoip2.record.Subdivision; @SideEffectFree @SupportsBatching @Tags({"geo", "enrich", "ip", "maxmind"}) +@InputRequirement(Requirement.INPUT_REQUIRED) @CapabilityDescription("Looks up geolocation information for an IP address and adds the geo information to FlowFile attributes. The " + "geo data is provided as a MaxMind database. The attribute that contains the IP address to lookup is provided by the " + "'IP Address Attribute' property. If the name of the attribute provided is 'X', then the the attributes added by enrichment "
