NIFI-15180 Log initialization errors in ConsumeKinesis

Author: awelless

nifi-extension-bundles/nifi-aws-bundle/nifi-aws-kinesis/src/main/java/org/apache/nifi/processors/aws/kinesis/ConsumeKinesis.java

@@ -31,6 +31,7 @@
 import org.apache.nifi.components.PropertyDescriptor;
 import org.apache.nifi.controller.NodeTypeProvider;
 import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.logging.ComponentLog;
 import org.apache.nifi.processor.AbstractProcessor;
 import org.apache.nifi.processor.DataUnit;
 import org.apache.nifi.processor.ProcessContext;
@@ -66,6 +67,7 @@
 import software.amazon.kinesis.common.InitialPositionInStream;
 import software.amazon.kinesis.common.InitialPositionInStreamExtended;
 import software.amazon.kinesis.coordinator.Scheduler;
+import software.amazon.kinesis.coordinator.WorkerStateChangeListener;
 import software.amazon.kinesis.lifecycle.events.InitializationInput;
 import software.amazon.kinesis.lifecycle.events.LeaseLostInput;
 import software.amazon.kinesis.lifecycle.events.ProcessRecordsInput;
@@ -320,6 +322,7 @@ Ensure that the credentials provided have access to Kinesis, DynamoDB and (optio
     private volatile RecordBuffer.ForProcessor<Lease> recordBuffer;
 
     private volatile @Nullable ReaderRecordProcessor readerRecordProcessor;
+    private volatile SchedulerChangeListener errorListener;
 
     // An instance filed, so that it can be read in getRelationships.
     private volatile ProcessingStrategy processingStrategy = ProcessingStrategy.from(
@@ -387,6 +390,8 @@ public void setup(final ProcessContext context) {
         recordBuffer = memoryBoundRecordBuffer;
         final ShardRecordProcessorFactory recordProcessorFactory = () -> new ConsumeKinesisRecordProcessor(memoryBoundRecordBuffer);
 
+        errorListener = new SchedulerChangeListener(getLogger());
+
         final String applicationName = context.getProperty(APPLICATION_NAME).getValue();
         final String workerId = generateWorkerId();
         final ConfigsBuilder configsBuilder = new ConfigsBuilder(streamTracker, applicationName, kinesisClient, dynamoDbClient, cloudWatchClient, workerId, recordProcessorFactory);
@@ -395,7 +400,7 @@ public void setup(final ProcessContext context) {
 
         kinesisScheduler = new Scheduler(
                 configsBuilder.checkpointConfig(),
-                configsBuilder.coordinatorConfig(),
+                configsBuilder.coordinatorConfig().workerStateChangeListener(errorListener),
                 configsBuilder.leaseManagementConfig(),
                 configsBuilder.lifecycleConfig(),
                 configsBuilder.metricsConfig().metricsFactory(metricsFactory),
@@ -528,6 +533,7 @@ public void onStopped() {
         }
 
         recordBuffer = null;
+        errorListener = null;
         readerRecordProcessor = null;
     }
 
@@ -560,6 +566,12 @@ private void shutdownScheduler() {
 
     @Override
     public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
+        if (errorListener.holdsInitializationFailure()) {
+            getLogger().error("Failed to initialize the processor. Correct the error and restart the processor.", errorListener.failure());
+            context.yield();
+            return;
+        }
+
         final Optional<Lease> leaseAcquired = recordBuffer.acquireBufferLease();
 
         leaseAcquired.ifPresentOrElse(
@@ -684,6 +696,35 @@ public void shutdownRequested(final ShutdownRequestedInput shutdownRequestedInpu
         }
     }
 
+    private static final class SchedulerChangeListener implements WorkerStateChangeListener {
+
+        private final ComponentLog logger;
+
+        private volatile @Nullable Throwable initializationFailure;
+
+        SchedulerChangeListener(final ComponentLog logger) {
+            this.logger = logger;
+        }
+
+        @Override
+        public void onWorkerStateChange(final WorkerState newState) {
+            logger.debug("Kinesis Scheduler changed state to: {}", newState);
+        }
+
+        @Override
+        public void onAllInitializationAttemptsFailed(final Throwable e) {
+            initializationFailure = e;
+        }
+
+        boolean holdsInitializationFailure() {
+            return initializationFailure != null;
+        }
+
+        @Nullable Throwable failure() {
+            return initializationFailure;
+        }
+    }
+
     enum ProcessingStrategy implements DescribedValue {
         FLOW_FILE("Write one FlowFile for each consumed Kinesis Record"),
         RECORD("Write one FlowFile containing multiple consumed Kinesis Records processed with Record Reader and Record Writer");

nifi-extension-bundles/nifi-aws-bundle/nifi-aws-kinesis/src/test/java/org/apache/nifi/processors/aws/kinesis/ConsumeKinesisTest.java

@@ -17,19 +17,28 @@
 package org.apache.nifi.processors.aws.kinesis;
 
 import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processors.aws.credentials.provider.service.AWSCredentialsProviderControllerService;
+import org.apache.nifi.processors.aws.region.RegionUtil;
+import org.apache.nifi.reporting.InitializationException;
+import org.apache.nifi.util.LogMessage;
 import org.apache.nifi.util.TestRunner;
 import org.apache.nifi.util.TestRunners;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
 
+import java.util.List;
 import java.util.Set;
 
+import static java.util.concurrent.TimeUnit.MINUTES;
 import static org.apache.nifi.processors.aws.kinesis.ConsumeKinesis.PROCESSING_STRATEGY;
 import static org.apache.nifi.processors.aws.kinesis.ConsumeKinesis.ProcessingStrategy.FLOW_FILE;
 import static org.apache.nifi.processors.aws.kinesis.ConsumeKinesis.ProcessingStrategy.RECORD;
 import static org.apache.nifi.processors.aws.kinesis.ConsumeKinesis.REL_PARSE_FAILURE;
 import static org.apache.nifi.processors.aws.kinesis.ConsumeKinesis.REL_SUCCESS;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Timeout.ThreadMode.SEPARATE_THREAD;
 
 class ConsumeKinesisTest {
 
@@ -57,4 +66,61 @@ void getRelationshipsForRecordProcessingStrategy() {
 
         assertEquals(Set.of(REL_SUCCESS, REL_PARSE_FAILURE), relationships);
     }
+
+    @Test
+    @Timeout(value = 2, unit = MINUTES, threadMode = SEPARATE_THREAD)
+    void logInitializationErrors() {
+        // With dummy values KCL Scheduler initialization will fail.
+        setDummyValues(testRunner);
+
+        awaitErrorMessages(testRunner);
+
+        final List<LogMessage> errorMessages = testRunner.getLogger().getErrorMessages();
+
+        final boolean hasInitializationError = errorMessages.stream()
+                .map(LogMessage::getMsg)
+                .anyMatch(msg -> msg.contains("Failed to initialize the processor. Correct the error and restart the processor."));
+
+        assertTrue(hasInitializationError, "Logs should contain initialization error message");
+    }
+
+    private static void awaitErrorMessages(final TestRunner runner) {
+        runner.run(1, false, true);
+
+        while (runner.getLogger().getErrorMessages().isEmpty()) {
+            runner.run(1, false, false);
+
+            try {
+                Thread.sleep(100);
+            } catch (final InterruptedException e) {
+                Thread.currentThread().interrupt();
+                throw new RuntimeException("Interrupted while waiting for error logs", e);
+            }
+        }
+
+        runner.run(1, true, false);
+    }
+
+    private static void setDummyValues(final TestRunner runner) {
+        final AWSCredentialsProviderControllerService credentialsService = new AWSCredentialsProviderControllerService();
+        try {
+            runner.addControllerService("credentials", credentialsService);
+        } catch (final InitializationException e) {
+            throw new RuntimeException(e);
+        }
+        runner.setProperty(credentialsService, AWSCredentialsProviderControllerService.ACCESS_KEY_ID, "123");
+        runner.setProperty(credentialsService, AWSCredentialsProviderControllerService.SECRET_KEY, "123");
+        runner.enableControllerService(credentialsService);
+
+        runner.setProperty(ConsumeKinesis.AWS_CREDENTIALS_PROVIDER_SERVICE, "credentials");
+        runner.setProperty(ConsumeKinesis.STREAM_NAME, "stream");
+        runner.setProperty(ConsumeKinesis.APPLICATION_NAME, "application");
+        runner.setProperty(RegionUtil.REGION, "us-west-2");
+        runner.setProperty(ConsumeKinesis.INITIAL_STREAM_POSITION, ConsumeKinesis.InitialPosition.TRIM_HORIZON);
+        runner.setProperty(ConsumeKinesis.PROCESSING_STRATEGY, ConsumeKinesis.ProcessingStrategy.FLOW_FILE);
+
+        runner.setProperty(ConsumeKinesis.METRICS_PUBLISHING, ConsumeKinesis.MetricsPublishing.CLOUDWATCH);
+
+        runner.setProperty(ConsumeKinesis.MAX_BYTES_TO_BUFFER, "10 MB");
+    }
 }