NIFI-15488: Added significant number of debug log messages as well as some info/error for connector-related events (#10803)

Author: markap14

nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorNode.java

@@ -164,9 +164,11 @@ public void prepareForUpdate() throws FlowUpdateException {
                                             + "; it must be UPDATING.");
         }
 
+        logger.debug("Preparing {} for update", this);
         try (final NarCloseable ignored = NarCloseable.withComponentNarLoader(extensionManager, getConnector().getClass(), getIdentifier())) {
             getConnector().prepareForUpdate(workingFlowContext, activeFlowContext);
             stateTransition.setCurrentState(ConnectorState.UPDATING);
+            logger.debug("Successfully prepared {} for update", this);
         } catch (final Throwable t) {
             logger.error("Failed to prepare update for {}", this, t);
 
@@ -184,6 +186,7 @@ public void prepareForUpdate() throws FlowUpdateException {
     public void inheritConfiguration(final List<VersionedConfigurationStep> activeConfig, final List<VersionedConfigurationStep> workingConfig,
                 final Bundle flowContextBundle) throws FlowUpdateException {
 
+        logger.debug("Inheriting configuration for {}", this);
         final MutableConnectorConfigurationContext configurationContext = createConfigurationContext(activeConfig);
         final FrameworkFlowContext inheritContext = flowContextFactory.createWorkingFlowContext(identifier,
             connectorDetails.getComponentLog(), configurationContext, flowContextBundle);
@@ -196,6 +199,8 @@ public void inheritConfiguration(final List<VersionedConfigurationStep> activeCo
             final StepConfiguration stepConfig = createStepConfiguration(step);
             setConfiguration(step.getName(), stepConfig, true);
         }
+
+        logger.debug("Successfully inherited configuration for {}", this);
     }
 
     private StepConfiguration createStepConfiguration(final VersionedConfigurationStep step) {
@@ -280,6 +285,7 @@ private void applyUpdate(final FrameworkFlowContext contextToInherit) throws Flo
 
         stateTransition.setCurrentState(ConnectorState.UPDATED);
         stateTransition.setDesiredState(ConnectorState.UPDATED);
+        logger.info("Successfully applied update for {}", this);
     }
 
     private void destroyWorkingContext() {
@@ -306,6 +312,7 @@ public void abortUpdate(final Throwable cause) {
         try (final NarCloseable ignored = NarCloseable.withComponentNarLoader(extensionManager, getConnector().getClass(), getIdentifier())) {
             getConnector().abortUpdate(workingFlowContext, cause);
         }
+        logger.debug("Aborted update for {}", this);
     }
 
     @Override
@@ -325,6 +332,7 @@ private void setConfiguration(final String stepName, final StepConfiguration con
         try (final NarCloseable ignored = NarCloseable.withComponentNarLoader(extensionManager, connector.getClass(), getIdentifier())) {
             logger.debug("Notifying {} of configuration change for configuration step {}", this, stepName);
             connector.onConfigurationStepConfigured(stepName, workingFlowContext);
+            logger.debug("Successfully set configuration for step {} on {}", stepName, this);
         } catch (final FlowUpdateException e) {
             throw e;
         } catch (final Exception e) {
@@ -442,6 +450,7 @@ public Future<Void> stop(final FlowEngine scheduler) {
 
     @Override
     public Future<Void> drainFlowFiles() {
+        logger.debug("Draining FlowFiles for {}", this);
         requireStopped("drain FlowFiles", ConnectorState.DRAINING);
 
         try (final NarCloseable ignored = NarCloseable.withComponentNarLoader(extensionManager, connectorDetails.getConnector().getClass(), getIdentifier())) {
@@ -451,7 +460,12 @@ public Future<Void> drainFlowFiles() {
 
             final CompletableFuture<Void> stateUpdateFuture = drainFuture.whenComplete((result, failureCause) -> {
                 drainFutureRef.set(null);
-                logger.info("Successfully drained FlowFiles from {}; ensuring all components are stopped.", this);
+
+                if (failureCause == null) {
+                    logger.info("Successfully drained FlowFiles for {}", this);
+                } else {
+                    logger.error("Failed to drain FlowFiles for {}", this, failureCause);
+                }
 
                 try {
                     connectorDetails.getConnector().stop(activeFlowContext);
@@ -465,6 +479,7 @@ public Future<Void> drainFlowFiles() {
 
             return stateUpdateFuture;
         } catch (final Throwable t) {
+            logger.error("Failed to drain FlowFiles for {}", this, t);
             stateTransition.setCurrentState(ConnectorState.STOPPED);
             throw t;
         }
@@ -509,15 +524,25 @@ public void verifyCanPurgeFlowFiles() throws IllegalStateException {
 
     @Override
     public Future<Void> purgeFlowFiles(final String requestor) {
+        logger.debug("Purging FlowFiles for {}", this);
         requireStopped("purge FlowFiles", ConnectorState.PURGING);
 
         try {
             final String dropRequestId = UUID.randomUUID().toString();
             final DropFlowFileStatus status = activeFlowContext.getManagedProcessGroup().dropAllFlowFiles(dropRequestId, requestor);
             final CompletableFuture<Void> future = status.getCompletionFuture();
-            final CompletableFuture<Void> stateUpdateFuture = future.whenComplete((result, failureCause) -> stateTransition.setCurrentState(ConnectorState.STOPPED));
+            final CompletableFuture<Void> stateUpdateFuture = future.whenComplete((result, failureCause) -> {
+                stateTransition.setCurrentState(ConnectorState.STOPPED);
+
+                if (failureCause == null) {
+                    logger.info("Successfully purged FlowFiles for {}", this);
+                } else {
+                    logger.error("Failed to purge FlowFiles for {}", this, failureCause);
+                }
+            });
             return stateUpdateFuture;
         } catch (final Throwable t) {
+            logger.error("Failed to purge FlowFiles for {}", this, t);
             stateTransition.setCurrentState(ConnectorState.STOPPED);
             throw t;
         }
@@ -541,6 +566,7 @@ private void requireStopped(final String action, final ConnectorState newState)
     }
 
     private void stopComponent(final FlowEngine scheduler, final CompletableFuture<Void> stopCompleteFuture) {
+        logger.debug("Stopping component for {}", this);
         try (final NarCloseable ignored = NarCloseable.withComponentNarLoader(extensionManager, connectorDetails.getConnector().getClass(), getIdentifier())) {
             connectorDetails.getConnector().stop(activeFlowContext);
         } catch (final Exception e) {
@@ -551,6 +577,7 @@ private void stopComponent(final FlowEngine scheduler, final CompletableFuture<V
 
         stateTransition.setCurrentState(ConnectorState.STOPPED);
         stopCompleteFuture.complete(null);
+        logger.info("Successfully stopped {}", this);
 
         final ConnectorState desiredState = getDesiredState();
         if (desiredState == ConnectorState.RUNNING) {
@@ -560,6 +587,7 @@ private void stopComponent(final FlowEngine scheduler, final CompletableFuture<V
     }
 
     private void startComponent(final ScheduledExecutorService scheduler, final CompletableFuture<Void> startCompleteFuture) {
+        logger.debug("Starting component for {}", this);
         final ConnectorState desiredState = getDesiredState();
         if (desiredState != ConnectorState.RUNNING) {
             logger.info("Will not start {} because the desired state is no longer RUNNING but is now {}", this, desiredState);
@@ -576,6 +604,7 @@ private void startComponent(final ScheduledExecutorService scheduler, final Comp
 
         stateTransition.setCurrentState(ConnectorState.RUNNING);
         startCompleteFuture.complete(null);
+        logger.info("Successfully started {}", this);
     }
 
 
@@ -654,17 +683,20 @@ public List<DescribedValue> fetchAllowableValues(final String stepName, final St
 
     @Override
     public void initializeConnector(final FrameworkConnectorInitializationContext initializationContext) {
+        logger.debug("Initializing {}", this);
         this.initializationContext = initializationContext;
 
         try (NarCloseable ignored = NarCloseable.withComponentNarLoader(extensionManager, getConnector().getClass(), getIdentifier())) {
             getConnector().initialize(initializationContext);
         }
 
         recreateWorkingFlowContext();
+        logger.info("Successfully initialized {}", this);
     }
 
     @Override
     public void loadInitialFlow() throws FlowUpdateException {
+        logger.debug("Loading initial flow for {}", this);
         if (initializationContext == null) {
             throw new IllegalStateException("Cannot load initial flow because " + this + " has not been initialized yet.");
         }
@@ -733,6 +765,7 @@ public boolean isValidationPaused() {
 
     @Override
     public List<ConfigVerificationResult> verifyConfigurationStep(final String stepName, final StepConfiguration configurationOverrides) {
+        logger.debug("Verifying configuration step {} for {}", stepName, this);
         final List<SecretReference> invalidSecretRefs = new ArrayList<>();
         final List<AssetReference> invalidAssetRefs = new ArrayList<>();
         final Map<String, String> resolvedPropertyOverrides = resolvePropertyReferences(configurationOverrides, invalidSecretRefs, invalidAssetRefs);
@@ -790,6 +823,7 @@ public List<ConfigVerificationResult> verifyConfigurationStep(final String stepN
                 results.addAll(invalidConfigResults);
             }
 
+            logger.debug("Completed verification of configuration step {} for {}", stepName, this);
             return results;
         }
     }
@@ -895,6 +929,7 @@ private Optional<ConfigurationStep> getConfigurationStep(final String stepName)
 
     @Override
     public List<ConfigVerificationResult> verify() {
+        logger.debug("Verifying {}", this);
         final List<ConfigVerificationResult> results = new ArrayList<>();
 
         final ValidationState state = performValidation();
@@ -909,13 +944,15 @@ public List<ConfigVerificationResult> verify() {
                 .explanation("There are " + validationFailureExplanations.size() + " validation failures: " + validationFailureExplanations)
                 .build());
 
+            logger.debug("Completed verification for {} with validation failures", this);
             return results;
         }
 
         try (NarCloseable ignored = NarCloseable.withComponentNarLoader(extensionManager, getConnector().getClass(), getIdentifier())) {
             results.addAll(getConnector().verify(workingFlowContext));
         }
 
+        logger.debug("Completed verification for {}", this);
         return results;
     }
 
@@ -955,6 +992,7 @@ public FrameworkFlowContext getWorkingFlowContext() {
     @Override
     public void discardWorkingConfiguration() {
         recreateWorkingFlowContext();
+        logger.debug("Discarded working configuration for {}", this);
     }
 
     @Override
@@ -1192,6 +1230,7 @@ public Collection<ValidationResult> getValidationErrors() {
 
     @Override
     public ValidationState performValidation() {
+        logger.debug("Performing validation for {}", this);
         try (final NarCloseable ignored = NarCloseable.withComponentNarLoader(extensionManager, getConnector().getClass(), getIdentifier())) {
 
             final ConnectorValidationContext validationContext = createValidationContext(activeFlowContext);

nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorRepository.java

@@ -53,10 +53,12 @@ public class StandardConnectorRepository implements ConnectorRepository {
 
     @Override
     public void initialize(final ConnectorRepositoryInitializationContext context) {
+        logger.debug("Initializing ConnectorRepository");
         this.extensionManager = context.getExtensionManager();
         this.requestReplicator = context.getRequestReplicator();
         this.secretsManager = context.getSecretsManager();
         this.assetRepository = new StandardConnectorAssetRepository(context.getAssetManager());
+        logger.debug("Successfully initialized ConnectorRepository");
     }
 
     @Override
@@ -67,10 +69,12 @@ public void addConnector(final ConnectorNode connector) {
     @Override
     public void restoreConnector(final ConnectorNode connector) {
         addConnector(connector);
+        logger.debug("Successfully restored {}", connector);
     }
 
     @Override
     public void removeConnector(final String connectorId) {
+        logger.debug("Removing {}", connectorId);
         final ConnectorNode connectorNode = connectors.get(connectorId);
         if (connectorNode == null) {
             throw new IllegalStateException("No connector found with ID " + connectorId);
@@ -136,6 +140,7 @@ private void restartConnector(final ConnectorNode connector, final CompletableFu
 
     @Override
     public void applyUpdate(final ConnectorNode connector, final ConnectorUpdateContext context) throws FlowUpdateException {
+        logger.debug("Applying update to {}", connector);
         final ConnectorState initialDesiredState = connector.getDesiredState();
         logger.info("Applying update to Connector {}", connector);
 
@@ -155,6 +160,7 @@ public void applyUpdate(final ConnectorNode connector, final ConnectorUpdateCont
     }
 
     private void updateConnector(final ConnectorNode connector, final ConnectorState initialDesiredState, final ConnectorUpdateContext context) {
+        logger.debug("Updating {}", connector);
         try {
             // Perform whatever preparation is necessary for the update. Default implementation is to stop the connector.
             logger.debug("Preparing {} for update", connector);
@@ -179,17 +185,17 @@ private void updateConnector(final ConnectorNode connector, final ConnectorState
             // If the initial desired state was RUNNING, start the connector again. Otherwise, stop it.
             // We don't simply leave it be as the prepareForUpdate / update may have changed the state of some components.
             if (initialDesiredState == ConnectorState.RUNNING) {
-                logger.info("Connector {} has been successfully updated; starting Connector to resume initial state", connector);
+                logger.info("{} has been successfully updated; starting to resume initial state", connector);
                 connector.start(lifecycleExecutor);
             } else {
-                logger.info("Connector {} has been successfully updated; stopping Connector to resume initial state", connector);
+                logger.info("{} has been successfully updated; stopping to resume initial state", connector);
                 connector.stop(lifecycleExecutor);
             }
 
             // We've updated the state of the connector so save flow again
             context.saveFlow();
         } catch (final Exception e) {
-            logger.error("Failed to apply connector update for {}", connector, e);
+            logger.error("Failed to apply update for {}", connector, e);
             connector.abortUpdate(e);
         }
     }
@@ -268,18 +274,22 @@ private void collectReferencedAssetIds(final FrameworkFlowContext flowContext, f
     @Override
     public void configureConnector(final ConnectorNode connector, final String stepName, final StepConfiguration configuration) throws FlowUpdateException {
         connector.setConfiguration(stepName, configuration);
+        logger.info("Successfully configured {} for step {}", connector, stepName);
     }
 
     @Override
     public void inheritConfiguration(final ConnectorNode connector, final List<VersionedConfigurationStep> activeFlowConfiguration,
                 final List<VersionedConfigurationStep> workingFlowConfiguration, final Bundle flowContextBundle) throws FlowUpdateException {
 
+        logger.debug("Inheriting configuration for {}", connector);
         connector.transitionStateForUpdating();
         connector.prepareForUpdate();
 
         try {
             connector.inheritConfiguration(activeFlowConfiguration, workingFlowConfiguration, flowContextBundle);
+            logger.debug("Successfully inherited configuration for {}", connector);
         } catch (final Exception e) {
+            logger.error("Failed to inherit configuration for {}", connector, e);
             connector.abortUpdate(e);
             throw e;
         }