Package org.apache.nifi.processors.standard

Class MonitorActivity

java.lang.Object
org.apache.nifi.components.AbstractConfigurableComponent
org.apache.nifi.processor.AbstractSessionFactoryProcessor
org.apache.nifi.processor.AbstractProcessor
org.apache.nifi.processors.standard.MonitorActivity
All Implemented Interfaces:
org.apache.nifi.components.ConfigurableComponent, org.apache.nifi.processor.Processor
@SideEffectFree @TriggerSerially @TriggerWhenEmpty @InputRequirement(INPUT_REQUIRED) @Tags({"monitor","flow","active","inactive","activity","detection"}) @CapabilityDescription("Monitors the flow for activity and sends out an indicator when the flow has not had any data for some specified amount of time and again when the flow\'s activity is restored") @WritesAttribute(attribute="inactivityStartMillis",description="The time at which Inactivity began, in the form of milliseconds since Epoch") @WritesAttribute(attribute="inactivityDurationMillis",description="The number of milliseconds that the inactivity has spanned") @Stateful(scopes={CLUSTER,LOCAL}, description="MonitorActivity stores the last timestamp at each node as state, so that it can examine activity at cluster wide. If \'Copy Attribute\' is set to true, then flow file attributes are also persisted. In local scope, it stores last known activity timestamp if the flow is inactive.") public class MonitorActivity extends org.apache.nifi.processor.AbstractProcessor

  • Field Details

    • STATE_KEY_COMMON_FLOW_ACTIVITY_INFO

      public static final String STATE_KEY_COMMON_FLOW_ACTIVITY_INFO
      See Also:

    • STATE_KEY_LOCAL_FLOW_ACTIVITY_INFO

      public static final String STATE_KEY_LOCAL_FLOW_ACTIVITY_INFO
      See Also:

    • SCOPE_NODE

      public static final org.apache.nifi.components.AllowableValue SCOPE_NODE

    • SCOPE_CLUSTER

      public static final org.apache.nifi.components.AllowableValue SCOPE_CLUSTER

    • REPORT_NODE_ALL

      public static final org.apache.nifi.components.AllowableValue REPORT_NODE_ALL

    • REPORT_NODE_PRIMARY

      public static final org.apache.nifi.components.AllowableValue REPORT_NODE_PRIMARY

    • THRESHOLD

      public static final org.apache.nifi.components.PropertyDescriptor THRESHOLD

    • CONTINUALLY_SEND_MESSAGES

      public static final org.apache.nifi.components.PropertyDescriptor CONTINUALLY_SEND_MESSAGES

    • ACTIVITY_RESTORED_MESSAGE

      public static final org.apache.nifi.components.PropertyDescriptor ACTIVITY_RESTORED_MESSAGE

    • WAIT_FOR_ACTIVITY

      public static final org.apache.nifi.components.PropertyDescriptor WAIT_FOR_ACTIVITY

    • RESET_STATE_ON_RESTART

      public static final org.apache.nifi.components.PropertyDescriptor RESET_STATE_ON_RESTART

    • INACTIVITY_MESSAGE

      public static final org.apache.nifi.components.PropertyDescriptor INACTIVITY_MESSAGE

    • COPY_ATTRIBUTES

      public static final org.apache.nifi.components.PropertyDescriptor COPY_ATTRIBUTES

    • MONITORING_SCOPE

      public static final org.apache.nifi.components.PropertyDescriptor MONITORING_SCOPE

    • REPORTING_NODE

      public static final org.apache.nifi.components.PropertyDescriptor REPORTING_NODE

    • PROPERTY_DESCRIPTORS

      private static final List<org.apache.nifi.components.PropertyDescriptor> PROPERTY_DESCRIPTORS

    • REL_SUCCESS

      public static final org.apache.nifi.processor.Relationship REL_SUCCESS

    • REL_INACTIVE

      public static final org.apache.nifi.processor.Relationship REL_INACTIVE

    • REL_ACTIVITY_RESTORED

      public static final org.apache.nifi.processor.Relationship REL_ACTIVITY_RESTORED

    • RELATIONSHIPS

      private static final Set<org.apache.nifi.processor.Relationship> RELATIONSHIPS

    • connectedWhenLastTriggered

      private final AtomicBoolean connectedWhenLastTriggered

    • lastInactiveMessage

      private final AtomicLong lastInactiveMessage

    • inactivityStartMillis

      private final AtomicLong inactivityStartMillis

    • wasActive

      private final AtomicBoolean wasActive

    • localFlowActivityInfo

      private volatile MonitorActivity.LocalFlowActivityInfo localFlowActivityInfo

  • Constructor Details

    • MonitorActivity

      public MonitorActivity()

  • Method Details

    • getRelationships

      public Set<org.apache.nifi.processor.Relationship> getRelationships()
      Specified by:
      getRelationships in interface org.apache.nifi.processor.Processor
      Overrides:
      getRelationships in class org.apache.nifi.processor.AbstractSessionFactoryProcessor

    • getSupportedPropertyDescriptors

      protected List<org.apache.nifi.components.PropertyDescriptor> getSupportedPropertyDescriptors()
      Overrides:
      getSupportedPropertyDescriptors in class org.apache.nifi.components.AbstractConfigurableComponent

    • onScheduled

      @OnScheduled public void onScheduled(org.apache.nifi.processor.ProcessContext context)

    • onStopped

      @OnStopped public void onStopped(org.apache.nifi.processor.ProcessContext context)

    • onTrigger

      public void onTrigger(org.apache.nifi.processor.ProcessContext context, org.apache.nifi.processor.ProcessSession session)
      Specified by:
      onTrigger in class org.apache.nifi.processor.AbstractProcessor

    • nowMillis

      protected long nowMillis()

    • getStartupTime

      protected long getStartupTime()

    • getLastSuccessfulTransfer

      protected final long getLastSuccessfulTransfer()

    • tryLoadLastSuccessfulTransfer

      private String tryLoadLastSuccessfulTransfer(org.apache.nifi.processor.ProcessContext context)

    • synchronizeState

      private void synchronizeState(org.apache.nifi.processor.ProcessContext context)

    • onTriggerInactiveFlow

      private void onTriggerInactiveFlow(org.apache.nifi.processor.ProcessContext context, org.apache.nifi.processor.ProcessSession session, boolean isClusterScope, long lastActivity)

    • onTriggerActiveFlow

      private void onTriggerActiveFlow(org.apache.nifi.processor.ProcessContext context, org.apache.nifi.processor.ProcessSession session, boolean wasActive, boolean isClusterScope, long inactivityStartMillis)

    • setInactivityFlag

      private void setInactivityFlag(org.apache.nifi.components.state.StateManager stateManager)

    • clearInactivityFlag

      private void clearInactivityFlag(org.apache.nifi.components.state.StateManager stateManager)

    • isClusterScope

      private boolean isClusterScope(org.apache.nifi.processor.ProcessContext context, boolean logInvalidConfig)

    • shouldReportOnlyOnPrimary

      private boolean shouldReportOnlyOnPrimary(boolean isClusterScope, org.apache.nifi.processor.ProcessContext context)

    • isReconnectedToCluster

      private boolean isReconnectedToCluster(boolean isConnectedToCluster)
      Will return true when the last known state is "not connected" and the current state is "connected". This might happen when during last @OnTrigger the node was not connected, but currently it is (reconnection); or when the processor is triggered first time (initial connection).
      This second case is due to safety reasons: it is possible that during the first trigger the node is not connected to the cluster thus the default value of the #connected attribute is false and stays as false until it's proven otherwise.
      Parameters:
      isConnectedToCluster - Current state of the connection.
      Returns:
      The node connected between the last trigger and the current one.

    • shouldThisNodeReport

      private boolean shouldThisNodeReport(boolean isClusterScope, org.apache.nifi.processor.ProcessContext context)

    • sendInactivityMarker

      private void sendInactivityMarker(org.apache.nifi.processor.ProcessContext context, org.apache.nifi.processor.ProcessSession session, long inactivityStartMillis, org.apache.nifi.logging.ComponentLog logger)

    • sendActivationMarker

      private void sendActivationMarker(org.apache.nifi.processor.ProcessContext context, org.apache.nifi.processor.ProcessSession session, Map<String,String> attributes, long inactivityStartMillis, org.apache.nifi.logging.ComponentLog logger)