This is an automated email from the ASF dual-hosted git repository.
pvillard31 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new 851de8d838b NIFI-16546 Deprecated ConsumeKinesisStream for Removal
(#11889)
851de8d838b is described below
commit 851de8d838bdbfe33c179351c573ca98165d9a86
Author: David Handermann <[email protected]>
AuthorDate: Mon Nov 9 03:19:50 2026 -0600
NIFI-16546 Deprecated ConsumeKinesisStream for Removal (#11889)
- Removed KCL references from Logback configuration
- Removed Integration Test for ConsumeKinesisStream
- Removed unit tests for subcomponents of ConsumeKinesisStream
---
.../nifi/processors/aws/AbstractAwsProcessor.java | 6 +-
.../aws/kinesis/stream/ConsumeKinesisStream.java | 5 +
.../aws/kinesis/stream/PutKinesisStream.java | 2 +-
.../additionalDetails.md | 46 --
.../aws/kinesis/stream/ITConsumeKinesisStream.java | 149 ------
.../kinesis/stream/TestConsumeKinesisStream.java | 2 +-
.../pause/TestStandardRecordProcessorBlocker.java | 205 --------
.../record/TestAbstractKinesisRecordProcessor.java | 200 --------
.../record/TestKinesisRecordProcessorRaw.java | 245 ----------
.../record/TestKinesisRecordProcessorRecord.java | 530 ---------------------
.../src/main/resources/conf/logback.xml | 3 -
.../src/main/resources/conf/stateless-logback.xml | 3 -
.../resources/conf/clustered/node1/logback.xml | 3 -
.../resources/conf/clustered/node2/logback.xml | 3 -
.../src/test/resources/conf/default/logback.xml | 3 -
.../src/test/resources/conf/pythonic/logback.xml | 3 -
16 files changed, 9 insertions(+), 1399 deletions(-)
diff --git
a/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/AbstractAwsProcessor.java
b/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/AbstractAwsProcessor.java
index a0dd073dd13..1024b525d82 100644
---
a/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/AbstractAwsProcessor.java
+++
b/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/AbstractAwsProcessor.java
@@ -156,10 +156,8 @@ public abstract class AbstractAwsProcessor<C extends
AwsClient, B extends AwsCli
*/
protected abstract void configureHttpClient(AwsClientBuilder<?, ?>
clientBuilder, ProcessContext context);
- /*
- * Allow optional override of onTrigger with the ProcessSessionFactory
where required for AWS processors (e.g. ConsumeKinesisStream)
- *
- * @see AbstractProcessor
+ /**
+ * Allow optional override of onTrigger with the ProcessSessionFactory
where required for AWS processors
*/
@Override
public void onTrigger(final ProcessContext context, final
ProcessSessionFactory sessionFactory) throws ProcessException {
diff --git
a/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/java/org/apache/nifi/processors/aws/kinesis/stream/ConsumeKinesisStream.java
b/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/java/org/apache/nifi/processors/aws/kinesis/stream/ConsumeKinesisStream.java
index 54660873fb2..8afa3320b19 100644
---
a/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/java/org/apache/nifi/processors/aws/kinesis/stream/ConsumeKinesisStream.java
+++
b/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/java/org/apache/nifi/processors/aws/kinesis/stream/ConsumeKinesisStream.java
@@ -32,6 +32,7 @@ import
org.apache.nifi.annotation.behavior.TriggerWhenAnyDestinationAvailable;
import org.apache.nifi.annotation.behavior.WritesAttribute;
import org.apache.nifi.annotation.behavior.WritesAttributes;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.DeprecationNotice;
import org.apache.nifi.annotation.documentation.SeeAlso;
import org.apache.nifi.annotation.documentation.Tags;
import org.apache.nifi.annotation.lifecycle.OnScheduled;
@@ -160,6 +161,10 @@ import static
org.apache.nifi.processors.aws.region.RegionUtil.getRegion;
@SystemResourceConsideration(resource = SystemResource.NETWORK, description =
"Kinesis Client Library will continually poll for new Records, " +
"requesting up to a maximum number of Records/bytes per call. This can
result in sustained network usage.")
@SeeAlso(PutKinesisStream.class)
+@DeprecationNotice(
+ reason = "NIFI-16546: Depends on the AWS Kinesis Client Library
requiring buffering and locking outside of framework flow control",
+ classNames = {"org.apache.nifi.processors.aws.kinesis.ConsumeKinesis"}
+)
public class ConsumeKinesisStream extends
AbstractAwsAsyncProcessor<KinesisAsyncClient, KinesisAsyncClientBuilder> {
private static final int INITIAL_WINDOW_SIZE_BYTES = 512 * 1024; // 512 KB
- suggested by KinesisClientUtil
diff --git
a/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/java/org/apache/nifi/processors/aws/kinesis/stream/PutKinesisStream.java
b/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/java/org/apache/nifi/processors/aws/kinesis/stream/PutKinesisStream.java
index b9aeb2a03ac..6f42d6b87c9 100644
---
a/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/java/org/apache/nifi/processors/aws/kinesis/stream/PutKinesisStream.java
+++
b/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/java/org/apache/nifi/processors/aws/kinesis/stream/PutKinesisStream.java
@@ -63,7 +63,7 @@ import static
org.apache.nifi.processors.aws.region.RegionUtil.REGION;
@WritesAttribute(attribute = "aws.kinesis.error.code", description =
"Error code for the message when posting to AWS Kinesis"),
@WritesAttribute(attribute = "aws.kinesis.sequence.number", description =
"Sequence number for the message when posting to AWS Kinesis"),
@WritesAttribute(attribute = "aws.kinesis.shard.id", description = "Shard
id of the message posted to AWS Kinesis")})
-@SeeAlso(ConsumeKinesisStream.class)
+@SeeAlso(classNames =
{"org.apache.nifi.processors.aws.kinesis.ConsumeKinesis"})
public class PutKinesisStream extends AbstractAwsSyncProcessor<KinesisClient,
KinesisClientBuilder> {
/**
* Kinesis put record response error message
diff --git
a/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/resources/docs/org.apache.nifi.processors.aws.kinesis.stream.ConsumeKinesisStream/additionalDetails.md
b/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/resources/docs/org.apache.nifi.processors.aws.kinesis.stream.ConsumeKinesisStream/additionalDetails.md
deleted file mode 100644
index 3f310ec0bb0..00000000000
---
a/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/resources/docs/org.apache.nifi.processors.aws.kinesis.stream.ConsumeKinesisStream/additionalDetails.md
+++ /dev/null
@@ -1,46 +0,0 @@
-<!--
- Licensed to the Apache Software Foundation (ASF) under one or more
- contributor license agreements. See the NOTICE file distributed with
- this work for additional information regarding copyright ownership.
- The ASF licenses this file to You under the Apache License, Version 2.0
- (the "License"); you may not use this file except in compliance with
- the License. You may obtain a copy of the License at
- http://www.apache.org/licenses/LICENSE-2.0
- Unless required by applicable law or agreed to in writing, software
- distributed under the License is distributed on an "AS IS" BASIS,
- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- See the License for the specific language governing permissions and
- limitations under the License.
--->
-
-# ConsumeKinesisStream
-
-## Streaming Versus Batch Processing
-
-ConsumeKinesisStream retrieves all Kinesis Records that it encounters in the
configured Kinesis Stream. There are two
-common, broadly defined use cases.
-
-### Per-Message Use Case
-
-By default, the Processor will create a separate FlowFile for each Kinesis
Record (message) in the Stream and add
-attributes for shard id, sequence number, etc.
-
-### Per-Batch Use Case
-
-Another common use case is the desire to process all Kinesis Records retrieved
from the Stream in a batch as a single
-FlowFile.
-
-The ConsumeKinesisStream Processor can optionally be configured with a Record
Reader and Record Writer. When a Record
-Reader and Record Writer are configured, a single FlowFile will be created
that will contain a Record for each Record
-within the batch of Kinesis Records (messages), instead of a separate FlowFile
per Kinesis Record.
-
-The FlowFiles emitted in this mode will include the standard `record.*`
attributes along with the same Kinesis Shard ID,
-Sequence Number and Approximate Arrival Timestamp; but the values will relate
to the **last** Kinesis Record that was
-processed in the batch of messages constituting the content of the FlowFile.
-
-Once a Record Writer is set the Output Strategy can be set to `Use Wrapper` or
`Use Content`. When `Use Wrapper` is
-picked the original content of the Kinesis Record will be wrapped under
`value` key and an additional `metadata`
-key will be populated with Stream Name, Shard ID, Partition Key, Sequence
Number, Sub Sequence Number, Approximate
-Arrival Timestamp and a Sharded Sequence Number that is a concatenation of
Sequence and Sub Sequence Numbers that allows
-a single comparison record ordering.
-When `Use Content` is picked the original content of the Kinesis Record will
be used as is.
diff --git
a/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/test/java/org/apache/nifi/processors/aws/kinesis/stream/ITConsumeKinesisStream.java
b/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/test/java/org/apache/nifi/processors/aws/kinesis/stream/ITConsumeKinesisStream.java
deleted file mode 100644
index 1c8786bc85a..00000000000
---
a/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/test/java/org/apache/nifi/processors/aws/kinesis/stream/ITConsumeKinesisStream.java
+++ /dev/null
@@ -1,149 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.nifi.processors.aws.kinesis.stream;
-
-import
org.apache.nifi.processors.aws.kinesis.stream.record.AbstractKinesisRecordProcessor;
-import org.apache.nifi.util.MockFlowFile;
-import org.apache.nifi.util.TestRunner;
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.Test;
-import software.amazon.awssdk.core.SdkBytes;
-import software.amazon.awssdk.services.dynamodb.DynamoDbClient;
-import software.amazon.awssdk.services.dynamodb.model.DeleteTableRequest;
-import software.amazon.awssdk.services.dynamodb.model.ListTablesResponse;
-import software.amazon.awssdk.services.kinesis.KinesisClient;
-import software.amazon.awssdk.services.kinesis.model.DeleteStreamRequest;
-import software.amazon.awssdk.services.kinesis.model.ListStreamsResponse;
-import software.amazon.awssdk.services.kinesis.model.PutRecordRequest;
-import software.amazon.kinesis.common.InitialPositionInStream;
-
-import java.io.IOException;
-import java.nio.charset.Charset;
-import java.util.List;
-import java.util.concurrent.Executors;
-
-public abstract class ITConsumeKinesisStream {
-
- static final String KINESIS_STREAM_NAME = "test-stream";
- static final String APPLICATION_NAME = "test-application";
-
- protected TestRunner runner;
-
- KinesisClient kinesis;
- DynamoDbClient dynamoDB;
-
- @Test
- public void readHorizon() throws InterruptedException, IOException {
- String partitionKey = "1";
-
- final PutRecordRequest request = PutRecordRequest.builder()
- .streamName(KINESIS_STREAM_NAME)
- .data(SdkBytes.fromString("horizon", Charset.defaultCharset()))
- .partitionKey(partitionKey)
- .build();
- kinesis.putRecord(request);
-
- startKCL(runner, InitialPositionInStream.TRIM_HORIZON);
-
- runner.assertAllFlowFilesTransferred(ConsumeKinesisStream.REL_SUCCESS,
1);
-
- final List<MockFlowFile> ffs =
runner.getFlowFilesForRelationship(PutKinesisStream.REL_SUCCESS);
- final MockFlowFile out = ffs.iterator().next();
-
-
out.assertAttributeEquals(AbstractKinesisRecordProcessor.AWS_KINESIS_PARTITION_KEY,
partitionKey);
-
- out.assertContentEquals("horizon".getBytes());
- }
-
- @Test
- public void readLatest() throws InterruptedException, IOException {
- String partitionKey = "1";
-
- final PutRecordRequest request = PutRecordRequest.builder()
- .streamName(KINESIS_STREAM_NAME)
- .data(SdkBytes.fromString("horizon", Charset.defaultCharset()))
- .partitionKey(partitionKey)
- .build();
- kinesis.putRecord(request);
-
- startKCL(runner, InitialPositionInStream.LATEST);
-
- final PutRecordRequest request2 = PutRecordRequest.builder()
- .streamName(KINESIS_STREAM_NAME)
- .data(SdkBytes.fromString("latest", Charset.defaultCharset()))
- .partitionKey(partitionKey)
- .build();
- kinesis.putRecord(request2);
-
- waitForKCLToProcessTheLatestMessage();
-
- runner.assertAllFlowFilesTransferred(ConsumeKinesisStream.REL_SUCCESS,
1);
-
- final List<MockFlowFile> ffs =
runner.getFlowFilesForRelationship(PutKinesisStream.REL_SUCCESS);
- final MockFlowFile out = ffs.iterator().next();
-
-
out.assertAttributeEquals(AbstractKinesisRecordProcessor.AWS_KINESIS_PARTITION_KEY,
partitionKey);
-
- out.assertContentEquals("latest".getBytes());
- }
-
- private void waitForKCLToProcessTheLatestMessage() throws
InterruptedException {
- Thread.sleep(10_000);
- }
-
- private void startKCL(TestRunner runner, InitialPositionInStream
initialPositionInStream) throws InterruptedException {
- runner.setProperty(ConsumeKinesisStream.INITIAL_STREAM_POSITION,
initialPositionInStream.name());
-
- runner.assertValid();
-
- Executors.newSingleThreadScheduledExecutor().submit((Runnable)
runner::run);
-
- Thread.sleep(30_000);
- }
-
- @AfterEach
- public void tearDown() throws InterruptedException {
- cleanupKinesis();
- cleanupDynamoDB();
-
- runner = null;
-
- Thread.sleep(2_000);
- }
-
- private void cleanupDynamoDB() {
- if (dynamoDB != null) {
- ListTablesResponse tableResults = dynamoDB.listTables();
- List<String> tableName = tableResults.tableNames();
- if (tableName.contains(APPLICATION_NAME)) {
-
dynamoDB.deleteTable(DeleteTableRequest.builder().tableName(APPLICATION_NAME).build());
- }
- }
- dynamoDB = null;
- }
-
- private void cleanupKinesis() {
- if (kinesis != null) {
- final ListStreamsResponse streamsResult = kinesis.listStreams();
- List<String> streamNames = streamsResult.streamNames();
- if (streamNames.contains(KINESIS_STREAM_NAME)) {
-
kinesis.deleteStream(DeleteStreamRequest.builder().streamName(KINESIS_STREAM_NAME).build());
- }
- }
- kinesis = null;
- }
-}
diff --git
a/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/test/java/org/apache/nifi/processors/aws/kinesis/stream/TestConsumeKinesisStream.java
b/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/test/java/org/apache/nifi/processors/aws/kinesis/stream/TestConsumeKinesisStream.java
index e937c9b786c..40e0758efaf 100644
---
a/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/test/java/org/apache/nifi/processors/aws/kinesis/stream/TestConsumeKinesisStream.java
+++
b/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/test/java/org/apache/nifi/processors/aws/kinesis/stream/TestConsumeKinesisStream.java
@@ -316,7 +316,7 @@ public class TestConsumeKinesisStream {
/*
* Trigger a run of the ConsumeKinesisStream processor, but expect the KCL
Worker to fail (it needs connections to AWS resources)
- * Assert that our code is being called by checking log output. The
ITConsumeKinesisStream integration tests prove actual AWS connectivity
+ * Assert that our code is being called by checking log output.
*/
private void runWorker(final boolean withCredentials, final boolean
waitForFailure) throws UnknownHostException, InitializationException,
InterruptedException {
final TestRunner mockConsumeKinesisStreamRunner =
TestRunners.newTestRunner(MockConsumeKinesisStream.class);
diff --git
a/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/test/java/org/apache/nifi/processors/aws/kinesis/stream/pause/TestStandardRecordProcessorBlocker.java
b/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/test/java/org/apache/nifi/processors/aws/kinesis/stream/pause/TestStandardRecordProcessorBlocker.java
deleted file mode 100644
index 592aecacf6f..00000000000
---
a/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/test/java/org/apache/nifi/processors/aws/kinesis/stream/pause/TestStandardRecordProcessorBlocker.java
+++ /dev/null
@@ -1,205 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.nifi.processors.aws.kinesis.stream.pause;
-
-import org.junit.jupiter.api.Nested;
-import org.junit.jupiter.api.Test;
-
-import java.time.Duration;
-import java.util.concurrent.atomic.AtomicLong;
-import java.util.function.Supplier;
-
-import static org.junit.jupiter.api.Assertions.assertTrue;
-
-public class TestStandardRecordProcessorBlocker {
-
- @Nested
- class ExplicitBlock {
-
- @Test
- public void testBlockAndUnblock() {
- final TestThreadInspector blockerInspector = new
TestThreadInspector();
-
- final StandardRecordProcessorBlocker recordProcessorBlocker =
StandardRecordProcessorBlocker.create();
-
- recordProcessorBlocker.block();
- final Thread thread = new
Thread(createRunnableWithInspector(recordProcessorBlocker, blockerInspector));
- thread.start();
-
- blockerInspector.awaitBlockAwaited();
- assertTrue(thread.isAlive());
-
- blockerInspector.awaitThreadParked(thread);
- recordProcessorBlocker.unblock();
- blockerInspector.awaitBlockExited();
- }
-
- @Test
- public void testNoBlock() {
- final TestThreadInspector blockerInspector = new
TestThreadInspector();
-
- final StandardRecordProcessorBlocker recordProcessorBlocker =
StandardRecordProcessorBlocker.create();
-
- recordProcessorBlocker.unblock();
- final Thread thread = new
Thread(createRunnableWithInspector(recordProcessorBlocker, blockerInspector));
- thread.start();
-
- blockerInspector.awaitBlockExited();
- }
-
- @Test
- public void testBlock() {
- final TestThreadInspector blockerInspector = new
TestThreadInspector();
-
- final StandardRecordProcessorBlocker recordProcessorBlocker =
StandardRecordProcessorBlocker.create();
- recordProcessorBlocker.block();
- final Thread thread = new
Thread(createRunnableWithInspector(recordProcessorBlocker, blockerInspector));
- thread.start();
-
- blockerInspector.awaitBlockAwaited();
- assertTrue(thread.isAlive());
-
- recordProcessorBlocker.unblock();
- }
- }
-
- @Nested
- class TimeoutBlock {
-
- final AtomicLong currentTimeMillis = new AtomicLong();
-
- @Test
- public void testBlockAndUnblock() {
- final TestThreadInspector blockerInspector = new
TestThreadInspector();
-
- final StandardRecordProcessorBlocker recordProcessorBlocker = new
StandardRecordProcessorBlocker(currentTimeMillis::get);
-
- currentTimeMillis.set(0L);
- recordProcessorBlocker.unblock();
-
currentTimeMillis.addAndGet(StandardRecordProcessorBlocker.BLOCK_AFTER_DURATION.toMillis()
+ 1);
- final Thread thread = new
Thread(createRunnableWithInspector(recordProcessorBlocker, blockerInspector));
- thread.start();
-
- blockerInspector.awaitBlockAwaited();
- assertTrue(thread.isAlive());
-
- blockerInspector.awaitThreadParked(thread);
- recordProcessorBlocker.unblock();
- blockerInspector.awaitBlockExited();
- }
-
- @Test
- public void testNoBlock() {
- final TestThreadInspector blockerInspector = new
TestThreadInspector();
-
- final StandardRecordProcessorBlocker recordProcessorBlocker = new
StandardRecordProcessorBlocker(currentTimeMillis::get);
-
- currentTimeMillis.set(0L);
- recordProcessorBlocker.unblock();
-
currentTimeMillis.addAndGet(StandardRecordProcessorBlocker.BLOCK_AFTER_DURATION.toMillis());
- final Thread thread = new
Thread(createRunnableWithInspector(recordProcessorBlocker, blockerInspector));
- thread.start();
-
- blockerInspector.awaitBlockExited();
- }
-
- @Test
- public void testNoBlockAfterClose() {
- final TestThreadInspector blockerInspector = new
TestThreadInspector();
-
- final StandardRecordProcessorBlocker recordProcessorBlocker = new
StandardRecordProcessorBlocker(currentTimeMillis::get);
-
- recordProcessorBlocker.unblock();
- recordProcessorBlocker.unblockIndefinitely();
- currentTimeMillis.set(Long.MAX_VALUE);
- final Thread thread = new
Thread(createRunnableWithInspector(recordProcessorBlocker, blockerInspector));
- thread.start();
-
- blockerInspector.awaitBlockExited();
- }
-
- @Test
- public void testBlock() {
- final TestThreadInspector blockerInspector = new
TestThreadInspector();
-
- final StandardRecordProcessorBlocker recordProcessorBlocker = new
StandardRecordProcessorBlocker(currentTimeMillis::get);
- currentTimeMillis.set(0L);
- recordProcessorBlocker.unblock();
-
currentTimeMillis.addAndGet(StandardRecordProcessorBlocker.BLOCK_AFTER_DURATION.toMillis()
+ 1);
- final Thread thread = new
Thread(createRunnableWithInspector(recordProcessorBlocker, blockerInspector));
- thread.start();
-
- blockerInspector.awaitBlockAwaited();
- assertTrue(thread.isAlive());
-
- recordProcessorBlocker.unblock();
- }
- }
-
- private static Runnable createRunnableWithInspector(final
StandardRecordProcessorBlocker recordProcessorBlocker, TestThreadInspector
blockerInspector) {
- return () -> {
- try {
- blockerInspector.onBlockAwaited();
- recordProcessorBlocker.await();
- } catch (final InterruptedException e) {
- throw new RuntimeException(e);
- } finally {
- blockerInspector.onBlockExited();
- }
- };
- }
-
- private static class TestThreadInspector {
- private static final Duration BUSY_WAIT_MAX_DURATION =
Duration.ofSeconds(5);
- private volatile boolean blockAwaited = false;
- private volatile boolean blockExited = false;
-
- public void onBlockAwaited() {
- blockAwaited = true;
- }
-
- public void onBlockExited() {
- blockExited = true;
- }
-
- public void awaitBlockAwaited() {
- busyWait(() -> !blockAwaited);
- }
-
- public void awaitBlockExited() {
- busyWait(() -> !blockExited);
- }
-
- public void awaitThreadParked(final Thread thread) {
- busyWait(() -> thread.getState() != Thread.State.WAITING);
- }
-
- private void busyWait(final Supplier<Boolean> condition) {
- final long maxWait = System.currentTimeMillis() +
BUSY_WAIT_MAX_DURATION.toMillis();
- do {
- if (System.currentTimeMillis() > maxWait) {
- throw new RuntimeException("Timed out waiting for
condition");
- }
- try {
- Thread.sleep(10L); // Sleep to avoid busy waiting too
aggressively
- } catch (final InterruptedException e) {
- throw new RuntimeException("Thread interrupted while
waiting for condition", e);
- }
- } while (condition.get());
- }
- }
-}
diff --git
a/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/test/java/org/apache/nifi/processors/aws/kinesis/stream/record/TestAbstractKinesisRecordProcessor.java
b/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/test/java/org/apache/nifi/processors/aws/kinesis/stream/record/TestAbstractKinesisRecordProcessor.java
deleted file mode 100644
index c6fca27f008..00000000000
---
a/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/test/java/org/apache/nifi/processors/aws/kinesis/stream/record/TestAbstractKinesisRecordProcessor.java
+++ /dev/null
@@ -1,200 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.nifi.processors.aws.kinesis.stream.record;
-
-import org.apache.nifi.processor.ProcessSessionFactory;
-import org.apache.nifi.processors.aws.kinesis.stream.ConsumeKinesisStream;
-import
org.apache.nifi.processors.aws.kinesis.stream.pause.StandardRecordProcessorBlocker;
-import org.apache.nifi.util.MockProcessSession;
-import org.apache.nifi.util.SharedSessionState;
-import org.apache.nifi.util.TestRunner;
-import org.apache.nifi.util.TestRunners;
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import software.amazon.kinesis.common.InitialPositionInStream;
-import software.amazon.kinesis.exceptions.InvalidStateException;
-import software.amazon.kinesis.exceptions.ShutdownException;
-import software.amazon.kinesis.exceptions.ThrottlingException;
-import software.amazon.kinesis.lifecycle.events.InitializationInput;
-import software.amazon.kinesis.lifecycle.events.ShutdownRequestedInput;
-import software.amazon.kinesis.processor.RecordProcessorCheckpointer;
-import software.amazon.kinesis.retrieval.KinesisClientRecord;
-import software.amazon.kinesis.retrieval.kpl.ExtendedSequenceNumber;
-
-import java.time.format.DateTimeFormatter;
-import java.util.concurrent.atomic.AtomicLong;
-
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertFalse;
-import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.mockito.Mockito.doThrow;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.reset;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.verifyNoMoreInteractions;
-import static org.mockito.Mockito.when;
-
-public class TestAbstractKinesisRecordProcessor {
- private static final DateTimeFormatter DATE_TIME_FORMATTER =
DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSS");
- private static final StandardRecordProcessorBlocker
NOOP_RECORD_PROCESSOR_BLOCKER = StandardRecordProcessorBlocker.create();
-
- private final TestRunner runner =
TestRunners.newTestRunner(ConsumeKinesisStream.class);
-
- private final ProcessSessionFactory processSessionFactory =
mock(ProcessSessionFactory.class);
-
- private final MockProcessSession session = new MockProcessSession(new
SharedSessionState(runner.getProcessor(), new AtomicLong(0)),
runner.getProcessor());
- private AbstractKinesisRecordProcessor fixture;
- private final RecordProcessorCheckpointer checkpointer =
mock(RecordProcessorCheckpointer.class);
-
- private final KinesisClientRecord kinesisRecord =
mock(KinesisClientRecord.class);
-
- @BeforeEach
- public void setUp() {
- NOOP_RECORD_PROCESSOR_BLOCKER.unblockIndefinitely();
- when(processSessionFactory.createSession()).thenReturn(session);
-
- // default test fixture will try operations twice with very little
wait in between
- fixture = new AbstractKinesisRecordProcessor(processSessionFactory,
runner.getLogger(), "kinesis-test",
- "endpoint-prefix", null, 10_000L, 1L, 2, DATE_TIME_FORMATTER,
NOOP_RECORD_PROCESSOR_BLOCKER) {
- @Override
- void processRecord(KinesisClientRecord kinesisRecord,
BatchProcessingContext batchProcessingContext) {
- // intentionally blank
- }
- };
- }
-
- @AfterEach
- public void tearDown() {
- verifyNoMoreInteractions(checkpointer, kinesisRecord,
processSessionFactory);
- reset(checkpointer, kinesisRecord, processSessionFactory);
- }
-
- @Test
- public void testInitialisation() {
- final ExtendedSequenceNumber esn = new
ExtendedSequenceNumber(InitialPositionInStream.AT_TIMESTAMP.toString(), 123L);
- final InitializationInput initializationInput =
InitializationInput.builder()
- .extendedSequenceNumber(esn)
- .shardId("shard-id")
- .build();
-
- fixture.initialize(initializationInput);
-
- assertTrue(fixture.getNextCheckpointTimeInMillis() >
System.currentTimeMillis());
- assertEquals("shard-id", fixture.getKinesisShardId());
- }
-
- @Test
- public void testInitialisationWithPendingCheckpoint() {
- final ExtendedSequenceNumber esn = new
ExtendedSequenceNumber(InitialPositionInStream.AT_TIMESTAMP.toString(), 123L);
- final ExtendedSequenceNumber prev = new
ExtendedSequenceNumber(InitialPositionInStream.LATEST.toString(), 456L);
- final InitializationInput initializationInput =
InitializationInput.builder()
- .extendedSequenceNumber(esn)
- .pendingCheckpointSequenceNumber(prev)
- .shardId("shard-id")
- .build();
-
- fixture.initialize(initializationInput);
-
- assertTrue(fixture.getNextCheckpointTimeInMillis() >
System.currentTimeMillis());
- assertEquals("shard-id", fixture.getKinesisShardId());
- }
-
- @Test
- public void testShutdown() throws InvalidStateException, ShutdownException
{
- final ShutdownRequestedInput shutdownInput =
ShutdownRequestedInput.builder()
- .checkpointer(checkpointer)
- .build();
-
- fixture.setKinesisShardId("test-shard");
- fixture.shutdownRequested(shutdownInput);
-
- verify(checkpointer, times(1)).checkpoint();
- }
-
- @Test
- public void testShutdownWithThrottlingFailures() throws
InvalidStateException, ShutdownException {
- final ShutdownRequestedInput shutdownInput =
ShutdownRequestedInput.builder()
- .checkpointer(checkpointer)
- .build();
-
- doThrow(new
ThrottlingException("throttled")).when(checkpointer).checkpoint();
-
- fixture.shutdownRequested(shutdownInput);
-
- verify(checkpointer, times(2)).checkpoint();
- }
-
- @Test
- public void testShutdownWithShutdownFailure() throws
InvalidStateException, ShutdownException {
- final ShutdownRequestedInput shutdownInput =
ShutdownRequestedInput.builder()
- .checkpointer(checkpointer)
- .build();
-
- doThrow(new
ShutdownException("shutdown")).when(checkpointer).checkpoint();
-
- fixture.shutdownRequested(shutdownInput);
-
- verify(checkpointer, times(1)).checkpoint();
- }
-
- @Test
- public void testShutdownWithInvalidStateFailure() throws
InvalidStateException, ShutdownException {
- final ShutdownRequestedInput shutdownInput =
ShutdownRequestedInput.builder()
- .checkpointer(checkpointer)
- .build();
-
- doThrow(new InvalidStateException("invalid
state")).when(checkpointer).checkpoint();
-
- fixture.shutdownRequested(shutdownInput);
-
- verify(checkpointer, times(1)).checkpoint();
-
- assertFalse(runner.getLogger().getErrorMessages().isEmpty());
- }
-
- @Test
- public void testShutdownTerminateRecordsNotProcessing() throws
InvalidStateException, ShutdownException {
- final ShutdownRequestedInput shutdownInput =
ShutdownRequestedInput.builder()
- .checkpointer(checkpointer)
- .build();
-
- fixture.setKinesisShardId("test-shard");
- fixture.setProcessingRecords(false);
- fixture.shutdownRequested(shutdownInput);
-
- verify(checkpointer, times(1)).checkpoint();
-
- assertTrue(runner.getLogger().getWarnMessages().isEmpty());
- }
-
- @Test
- public void testShutdownTerminateRecordsProcessing() throws
InvalidStateException, ShutdownException {
- final ShutdownRequestedInput shutdownInput =
ShutdownRequestedInput.builder()
- .checkpointer(checkpointer)
- .build();
-
- fixture.setKinesisShardId("test-shard");
- fixture.setProcessingRecords(true);
- fixture.shutdownRequested(shutdownInput);
-
- verify(checkpointer, times(1)).checkpoint();
-
- assertFalse(runner.getLogger().getWarnMessages().isEmpty());
- }
-}
diff --git
a/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/test/java/org/apache/nifi/processors/aws/kinesis/stream/record/TestKinesisRecordProcessorRaw.java
b/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/test/java/org/apache/nifi/processors/aws/kinesis/stream/record/TestKinesisRecordProcessorRaw.java
deleted file mode 100644
index fe4c68357c5..00000000000
---
a/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/test/java/org/apache/nifi/processors/aws/kinesis/stream/record/TestKinesisRecordProcessorRaw.java
+++ /dev/null
@@ -1,245 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.nifi.processors.aws.kinesis.stream.record;
-
-import org.apache.nifi.processor.ProcessSessionFactory;
-import org.apache.nifi.processors.aws.kinesis.stream.ConsumeKinesisStream;
-import
org.apache.nifi.processors.aws.kinesis.stream.pause.StandardRecordProcessorBlocker;
-import org.apache.nifi.util.MockFlowFile;
-import org.apache.nifi.util.MockProcessSession;
-import org.apache.nifi.util.SharedSessionState;
-import org.apache.nifi.util.TestRunner;
-import org.apache.nifi.util.TestRunners;
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.mockito.Mock;
-import software.amazon.kinesis.exceptions.InvalidStateException;
-import software.amazon.kinesis.exceptions.ShutdownException;
-import software.amazon.kinesis.lifecycle.events.ProcessRecordsInput;
-import software.amazon.kinesis.processor.RecordProcessorCheckpointer;
-import software.amazon.kinesis.retrieval.KinesisClientRecord;
-
-import java.nio.ByteBuffer;
-import java.nio.charset.StandardCharsets;
-import java.time.Instant;
-import java.time.ZoneId;
-import java.time.ZonedDateTime;
-import java.time.format.DateTimeFormatter;
-import java.time.temporal.ChronoUnit;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.Date;
-import java.util.List;
-import java.util.concurrent.atomic.AtomicLong;
-
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertFalse;
-import static org.junit.jupiter.api.Assertions.assertNull;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.reset;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.verifyNoMoreInteractions;
-import static org.mockito.Mockito.when;
-
-public class TestKinesisRecordProcessorRaw {
- private static final DateTimeFormatter DATE_TIME_FORMATTER =
DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSS");
- private static final StandardRecordProcessorBlocker
NOOP_RECORD_PROCESSOR_BLOCKER = StandardRecordProcessorBlocker.create();
-
- private final TestRunner runner =
TestRunners.newTestRunner(ConsumeKinesisStream.class);
-
- private final ProcessSessionFactory processSessionFactory =
mock(ProcessSessionFactory.class);
-
- private final SharedSessionState sharedState = new
SharedSessionState(runner.getProcessor(), new AtomicLong(0));
- private final MockProcessSession session = new
MockProcessSession(sharedState, runner.getProcessor());
-
- private AbstractKinesisRecordProcessor fixture;
-
- @Mock
- private final RecordProcessorCheckpointer checkpointer =
mock(RecordProcessorCheckpointer.class);
-
- @Mock
- private final KinesisClientRecord kinesisRecord =
mock(KinesisClientRecord.class);
-
- @BeforeEach
- public void setUp() {
- NOOP_RECORD_PROCESSOR_BLOCKER.unblockIndefinitely();
- // default test fixture will try operations twice with very little
wait in between
- fixture = new KinesisRecordProcessorRaw(processSessionFactory,
runner.getLogger(), "kinesis-test",
- "endpoint-prefix", null, 10_000L, 1L, 2, DATE_TIME_FORMATTER,
NOOP_RECORD_PROCESSOR_BLOCKER);
- }
-
- @AfterEach
- public void tearDown() {
- verifyNoMoreInteractions(checkpointer, kinesisRecord,
processSessionFactory);
- reset(checkpointer, kinesisRecord, processSessionFactory);
- }
-
- @Test
- public void testProcessRecordsEmpty() {
- final ProcessRecordsInput processRecordsInput =
ProcessRecordsInput.builder()
- .records(Collections.emptyList())
- .checkpointer(checkpointer)
- .cacheEntryTime(Instant.now().minus(1, ChronoUnit.MINUTES))
- .cacheExitTime(Instant.now().minus(1, ChronoUnit.SECONDS))
- .millisBehindLatest(100L)
- .build();
-
- // would checkpoint (but should skip because there are no records
processed)
- fixture.setNextCheckpointTimeInMillis(System.currentTimeMillis() -
10_000L);
-
- fixture.processRecords(processRecordsInput);
-
- session.assertTransferCount(ConsumeKinesisStream.REL_SUCCESS, 0);
- assertEquals(0, sharedState.getProvenanceEvents().size());
- session.assertNotCommitted();
- session.assertNotRolledBack();
- }
-
- @Test
- public void testProcessRecordsNoCheckpoint() {
- processMultipleRecordsAssertProvenance(false);
- }
-
- @Test
- public void testProcessRecordsWithEndpointOverride() {
- processMultipleRecordsAssertProvenance(true);
- }
-
- private void processMultipleRecordsAssertProvenance(final boolean
endpointOverridden) {
- final Date firstDate = Date.from(Instant.now().minus(1,
ChronoUnit.MINUTES));
- final Date secondDate = new Date();
-
- final ProcessRecordsInput processRecordsInput =
ProcessRecordsInput.builder()
- .records(Arrays.asList(
-
KinesisClientRecord.builder().approximateArrivalTimestamp(firstDate.toInstant())
- .partitionKey("partition-1")
- .sequenceNumber("1")
-
.data(ByteBuffer.wrap("record-1".getBytes(StandardCharsets.UTF_8)))
- .build(),
-
KinesisClientRecord.builder().approximateArrivalTimestamp(secondDate.toInstant())
- .partitionKey("partition-2")
- .sequenceNumber("2")
-
.data(ByteBuffer.wrap("record-2".getBytes(StandardCharsets.UTF_8)))
- .build(),
-
KinesisClientRecord.builder().approximateArrivalTimestamp(null)
- .partitionKey("partition-no-date")
- .sequenceNumber("no-date")
-
.data(ByteBuffer.wrap("record-no-date".getBytes(StandardCharsets.UTF_8)))
- .build()
- ))
- .checkpointer(checkpointer)
- .cacheEntryTime(null)
- .cacheExitTime(null)
- .millisBehindLatest(null)
- .build();
-
- final String transitUriPrefix = endpointOverridden ?
"https://another-endpoint.com:8443"; : "http://endpoint-prefix.amazonaws.com";;
- if (endpointOverridden) {
- fixture = new KinesisRecordProcessorRaw(processSessionFactory,
runner.getLogger(), "kinesis-test",
- "endpoint-prefix", "https://another-endpoint.com:8443";,
10_000L, 1L, 2, DATE_TIME_FORMATTER, NOOP_RECORD_PROCESSOR_BLOCKER);
- }
-
- // skip checkpoint
- fixture.setNextCheckpointTimeInMillis(System.currentTimeMillis() +
10_000L);
- fixture.setKinesisShardId("test-shard");
-
- when(processSessionFactory.createSession()).thenReturn(session);
- fixture.processRecords(processRecordsInput);
- verify(processSessionFactory, times(1)).createSession();
-
- session.assertTransferCount(ConsumeKinesisStream.REL_SUCCESS,
processRecordsInput.records().size());
- assertEquals(sharedState.getProvenanceEvents().size(),
processRecordsInput.records().size());
- assertEquals(sharedState.getProvenanceEvents().get(0).getTransitUri(),
String.format("%s/test-shard/partition-1#1", transitUriPrefix));
- assertEquals(sharedState.getProvenanceEvents().get(1).getTransitUri(),
String.format("%s/test-shard/partition-2#2", transitUriPrefix));
- assertEquals(sharedState.getProvenanceEvents().get(2).getTransitUri(),
String.format("%s/test-shard/partition-no-date#no-date", transitUriPrefix));
-
- final List<MockFlowFile> flowFiles =
session.getFlowFilesForRelationship(ConsumeKinesisStream.REL_SUCCESS);
- assertFlowFile(flowFiles.get(0), firstDate, "partition-1", "1",
"record-1");
- assertFlowFile(flowFiles.get(1), secondDate, "partition-2", "2",
"record-2");
- assertFlowFile(flowFiles.get(2), null, "partition-no-date", "no-date",
"record-no-date");
-
- session.assertCommitted();
- session.assertNotRolledBack();
- }
-
- @Test
- public void testProcessPoisonPillRecordWithCheckpoint() throws
ShutdownException, InvalidStateException {
- final ProcessRecordsInput processRecordsInput =
ProcessRecordsInput.builder()
- .records(Arrays.asList(
-
KinesisClientRecord.builder().approximateArrivalTimestamp(null)
- .partitionKey("partition-1")
- .sequenceNumber("1")
-
.data(ByteBuffer.wrap("record-1".getBytes(StandardCharsets.UTF_8)))
- .build(),
- kinesisRecord,
-
KinesisClientRecord.builder().approximateArrivalTimestamp(null)
- .partitionKey("partition-3")
- .sequenceNumber("3")
-
.data(ByteBuffer.wrap("record-3".getBytes(StandardCharsets.UTF_8)))
- .build()
- ))
- .checkpointer(checkpointer)
- .cacheEntryTime(Instant.now().minus(1, ChronoUnit.MINUTES))
- .cacheExitTime(Instant.now().minus(1, ChronoUnit.SECONDS))
- .millisBehindLatest(100L)
- .build();
-
- when(kinesisRecord.data()).thenThrow(new
IllegalStateException("illegal state"));
- when(kinesisRecord.toString()).thenReturn("poison-pill");
-
- fixture.setKinesisShardId("test-shard");
-
- when(processSessionFactory.createSession()).thenReturn(session);
- fixture.processRecords(processRecordsInput);
- verify(processSessionFactory, times(1)).createSession();
-
- // check non-poison pill records are output successfully
- session.assertTransferCount(ConsumeKinesisStream.REL_SUCCESS, 2);
- final List<MockFlowFile> flowFiles =
session.getFlowFilesForRelationship(ConsumeKinesisStream.REL_SUCCESS);
- assertFlowFile(flowFiles.get(0), null, "partition-1", "1", "record-1");
- assertFlowFile(flowFiles.get(1), null, "partition-3", "3", "record-3");
-
- // check the "poison pill" record was retried a 2nd time
- assertNull(verify(kinesisRecord, times(2)).partitionKey());
- assertNull(verify(kinesisRecord, times(2)).sequenceNumber());
- assertNull(verify(kinesisRecord,
times(2)).approximateArrivalTimestamp());
- assertNull(verify(kinesisRecord, times(2)).data());
- verify(checkpointer, times(1)).checkpoint();
-
- assertFalse(runner.getLogger().getErrorMessages().isEmpty());
- assertFalse(runner.getLogger().getErrorMessages().isEmpty());
-
- session.assertCommitted();
- session.assertNotRolledBack();
- }
-
- private void assertFlowFile(final MockFlowFile flowFile, final Date
approxTimestamp, final String partitionKey,
- final String sequenceNumber, final String
content) {
- if (approxTimestamp != null) {
-
flowFile.assertAttributeEquals(AbstractKinesisRecordProcessor.AWS_KINESIS_APPROXIMATE_ARRIVAL_TIMESTAMP,
-
DATE_TIME_FORMATTER.format(ZonedDateTime.ofInstant(approxTimestamp.toInstant(),
ZoneId.systemDefault())));
- } else {
-
flowFile.assertAttributeNotExists(AbstractKinesisRecordProcessor.AWS_KINESIS_APPROXIMATE_ARRIVAL_TIMESTAMP);
- }
-
flowFile.assertAttributeEquals(AbstractKinesisRecordProcessor.AWS_KINESIS_PARTITION_KEY,
partitionKey);
-
flowFile.assertAttributeEquals(AbstractKinesisRecordProcessor.AWS_KINESIS_SEQUENCE_NUMBER,
sequenceNumber);
-
flowFile.assertAttributeEquals(AbstractKinesisRecordProcessor.AWS_KINESIS_SHARD_ID,
"test-shard");
- flowFile.assertContentEquals(content);
- }
-}
diff --git
a/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/test/java/org/apache/nifi/processors/aws/kinesis/stream/record/TestKinesisRecordProcessorRecord.java
b/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/test/java/org/apache/nifi/processors/aws/kinesis/stream/record/TestKinesisRecordProcessorRecord.java
deleted file mode 100644
index 5b39e97411c..00000000000
---
a/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/test/java/org/apache/nifi/processors/aws/kinesis/stream/record/TestKinesisRecordProcessorRecord.java
+++ /dev/null
@@ -1,530 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.nifi.processors.aws.kinesis.stream.record;
-
-import org.apache.commons.lang3.tuple.Pair;
-import org.apache.nifi.flowfile.attributes.CoreAttributes;
-import org.apache.nifi.json.JsonRecordSetWriter;
-import org.apache.nifi.json.JsonTreeReader;
-import org.apache.nifi.processor.ProcessSessionFactory;
-import
org.apache.nifi.processors.aws.kinesis.property.SchemaDifferenceHandlingStrategy;
-import org.apache.nifi.processors.aws.kinesis.stream.ConsumeKinesisStream;
-import
org.apache.nifi.processors.aws.kinesis.stream.pause.StandardRecordProcessorBlocker;
-import
org.apache.nifi.processors.aws.kinesis.stream.record.converter.RecordConverterIdentity;
-import
org.apache.nifi.processors.aws.kinesis.stream.record.converter.RecordConverterWrapper;
-import org.apache.nifi.reporting.InitializationException;
-import org.apache.nifi.schema.access.SchemaAccessUtils;
-import org.apache.nifi.schema.inference.SchemaInferenceUtil;
-import org.apache.nifi.serialization.RecordReaderFactory;
-import org.apache.nifi.serialization.RecordSetWriterFactory;
-import org.apache.nifi.util.MockFlowFile;
-import org.apache.nifi.util.MockProcessSession;
-import org.apache.nifi.util.SharedSessionState;
-import org.apache.nifi.util.TestRunner;
-import org.apache.nifi.util.TestRunners;
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Nested;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.Arguments;
-import org.junit.jupiter.params.provider.EnumSource;
-import org.junit.jupiter.params.provider.MethodSource;
-import org.mockito.Mock;
-import software.amazon.kinesis.exceptions.InvalidStateException;
-import software.amazon.kinesis.exceptions.ShutdownException;
-import software.amazon.kinesis.lifecycle.events.ProcessRecordsInput;
-import software.amazon.kinesis.processor.RecordProcessorCheckpointer;
-import software.amazon.kinesis.retrieval.KinesisClientRecord;
-
-import java.nio.ByteBuffer;
-import java.nio.charset.StandardCharsets;
-import java.time.Instant;
-import java.time.ZoneId;
-import java.time.ZonedDateTime;
-import java.time.format.DateTimeFormatter;
-import java.time.temporal.ChronoUnit;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.Date;
-import java.util.List;
-import java.util.concurrent.atomic.AtomicLong;
-import java.util.stream.Stream;
-
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertFalse;
-import static org.junit.jupiter.api.Assertions.assertNull;
-import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.reset;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.verifyNoMoreInteractions;
-import static org.mockito.Mockito.when;
-
-public class TestKinesisRecordProcessorRecord {
- private static final DateTimeFormatter DATE_TIME_FORMATTER =
DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSS");
- private static final StandardRecordProcessorBlocker
NOOP_RECORD_PROCESSOR_BLOCKER = StandardRecordProcessorBlocker.create();
-
- private final TestRunner runner =
TestRunners.newTestRunner(ConsumeKinesisStream.class);
-
- private final ProcessSessionFactory processSessionFactory =
mock(ProcessSessionFactory.class);
-
- private final SharedSessionState sharedState = new
SharedSessionState(runner.getProcessor(), new AtomicLong(0));
- private final MockProcessSession session = new
MockProcessSession(sharedState, runner.getProcessor());
-
- private AbstractKinesisRecordProcessor fixture;
- private final RecordReaderFactory reader = new JsonTreeReader();
- private final RecordSetWriterFactory writer = new JsonRecordSetWriter();
-
- @Mock
- private final RecordProcessorCheckpointer checkpointer =
mock(RecordProcessorCheckpointer.class);
-
- @Mock
- private final KinesisClientRecord kinesisRecord =
mock(KinesisClientRecord.class);
-
- @BeforeEach
- public void setUp() throws InitializationException {
- NOOP_RECORD_PROCESSOR_BLOCKER.unblockIndefinitely();
-
- runner.addControllerService("record-reader", reader);
- runner.setProperty(reader, SchemaAccessUtils.SCHEMA_ACCESS_STRATEGY,
SchemaInferenceUtil.INFER_SCHEMA.getValue());
- runner.enableControllerService(reader);
- runner.setProperty(ConsumeKinesisStream.RECORD_READER,
"record-reader");
-
- runner.addControllerService("record-writer", writer);
- runner.setProperty(writer, SchemaAccessUtils.SCHEMA_ACCESS_STRATEGY,
SchemaAccessUtils.INHERIT_RECORD_SCHEMA.getValue());
- runner.setProperty(writer, JsonRecordSetWriter.OUTPUT_GROUPING,
"output-oneline");
- runner.enableControllerService(writer);
- runner.setProperty(ConsumeKinesisStream.RECORD_WRITER,
"record-writer");
- }
-
- private KinesisRecordProcessorRecord defaultFixtureWithStrategy(final
SchemaDifferenceHandlingStrategy strategy) {
- // default test fixture will try operations twice with very little
wait in between
- return new KinesisRecordProcessorRecord(processSessionFactory,
runner.getLogger(), "kinesis-test",
- "endpoint-prefix", null, 10_000L, 1L, 2, DATE_TIME_FORMATTER,
- reader, writer, new RecordConverterIdentity(),
NOOP_RECORD_PROCESSOR_BLOCKER, strategy);
- }
-
- @AfterEach
- public void tearDown() {
- verifyNoMoreInteractions(checkpointer, kinesisRecord,
processSessionFactory);
- reset(checkpointer, kinesisRecord, processSessionFactory);
- }
-
- @ParameterizedTest
- @EnumSource(SchemaDifferenceHandlingStrategy.class)
- public void testProcessRecordsEmpty(final SchemaDifferenceHandlingStrategy
strategy) {
- fixture = defaultFixtureWithStrategy(strategy);
- final ProcessRecordsInput processRecordsInput =
ProcessRecordsInput.builder()
- .records(Collections.emptyList())
- .checkpointer(checkpointer)
- .cacheEntryTime(Instant.now().minus(1, ChronoUnit.MINUTES))
- .cacheExitTime(Instant.now().minus(1, ChronoUnit.SECONDS))
- .millisBehindLatest(100L)
- .build();
-
- // would checkpoint (but should skip because there are no records
processed)
- fixture.setNextCheckpointTimeInMillis(System.currentTimeMillis() -
10_000L);
-
- fixture.processRecords(processRecordsInput);
-
- session.assertTransferCount(ConsumeKinesisStream.REL_SUCCESS, 0);
- assertTrue(sharedState.getProvenanceEvents().isEmpty());
- session.assertNotCommitted();
- session.assertNotRolledBack();
- }
-
- private static Stream<Arguments.ArgumentSet> testProcessRecordsArgs() {
- final List<Pair<String, Boolean>> endpointOverridden = List.of(
- Pair.of("Overriden Endpoint", true),
- Pair.of("Default Endpoint", false)
- );
- final List<Pair<String, Boolean>> usingWrapper = List.of(
- Pair.of("Using Schema Wrapper", true),
- Pair.of("Default Schema", false)
- );
- final List<Pair<String, SchemaDifferenceHandlingStrategy>>
schemaChangeStrategy = Arrays.stream(SchemaDifferenceHandlingStrategy.values())
- .map(strategy -> Pair.of(strategy.name(), strategy))
- .toList();
- return endpointOverridden.stream()
- .flatMap(endpoint -> schemaChangeStrategy.stream()
- .flatMap(strategy -> usingWrapper.stream()
- .map(wrapper -> Arguments.argumentSet(
- String.format("%s, %s, %s",
endpoint.getLeft(), wrapper.getLeft(), strategy.getLeft()),
- endpoint.getRight(),
- wrapper.getRight(),
- strategy.getRight()
- ))));
- }
-
- @ParameterizedTest
- @MethodSource("testProcessRecordsArgs")
- void processMultipleRecordsAssertProvenance(final boolean
endpointOverridden, final boolean useWrapper, final
SchemaDifferenceHandlingStrategy strategy) {
- fixture = defaultFixtureWithStrategy(strategy);
- final Instant referenceInstant =
Instant.parse("2021-01-01T00:00:00.000Z");
- final Date firstDate = Date.from(referenceInstant.minus(1,
ChronoUnit.MINUTES));
- final Date secondDate = Date.from(referenceInstant);
-
- final ProcessRecordsInput processRecordsInput =
ProcessRecordsInput.builder()
- .records(Arrays.asList(
-
KinesisClientRecord.builder().approximateArrivalTimestamp(firstDate.toInstant())
- .partitionKey("partition-1")
- .sequenceNumber("1")
-
.data(ByteBuffer.wrap("{\"record\":\"1\"}\n{\"record\":\"1b\"}".getBytes(StandardCharsets.UTF_8)))
- .build(),
-
KinesisClientRecord.builder().approximateArrivalTimestamp(null)
- .partitionKey("partition-no-date")
- .sequenceNumber("no-date")
-
.data(ByteBuffer.wrap("{\"record\":\"no-date\"}".getBytes(StandardCharsets.UTF_8)))
- .build(),
-
KinesisClientRecord.builder().approximateArrivalTimestamp(secondDate.toInstant())
- .partitionKey("partition-2")
- .sequenceNumber("2")
-
.data(ByteBuffer.wrap("{\"record\":\"2\"}".getBytes(StandardCharsets.UTF_8)))
- .build()
- ))
- .checkpointer(checkpointer)
- .cacheEntryTime(null)
- .cacheExitTime(null)
- .millisBehindLatest(null)
- .build();
-
- final String transitUriPrefix = endpointOverridden ?
"https://another-endpoint.com:8443"; : "http://endpoint-prefix.amazonaws.com";;
- fixture = new KinesisRecordProcessorRecord(processSessionFactory,
runner.getLogger(), "kinesis-test",
- "endpoint-prefix", transitUriPrefix, 10_000L, 1L, 2,
DATE_TIME_FORMATTER,
- reader, writer, useWrapper ? new RecordConverterWrapper() :
new RecordConverterIdentity(), NOOP_RECORD_PROCESSOR_BLOCKER, strategy);
-
- // skip checkpoint
- fixture.setNextCheckpointTimeInMillis(System.currentTimeMillis() +
10_000L);
- fixture.setKinesisShardId("another-shard");
-
- when(processSessionFactory.createSession()).thenReturn(session);
- fixture.processRecords(processRecordsInput);
- verify(processSessionFactory, times(1)).createSession();
-
- session.assertTransferCount(ConsumeKinesisStream.REL_SUCCESS, 1);
- assertEquals(1, sharedState.getProvenanceEvents().size());
- assertEquals(String.format("%s/another-shard", transitUriPrefix),
sharedState.getProvenanceEvents().getFirst().getTransitUri());
-
- final List<MockFlowFile> flowFiles =
session.getFlowFilesForRelationship(ConsumeKinesisStream.REL_SUCCESS);
- // 4 records in single output file, attributes equating to that of the
last record
- final String expectedContent = useWrapper ?
expectRecordContentWrapper() : expectRecordContentIdentity();
- assertFlowFile(flowFiles.getFirst(), secondDate, "partition-2", "2",
"another-shard", expectedContent, 4);
- session.assertTransferCount(ConsumeKinesisStream.REL_PARSE_FAILURE, 0);
-
- session.assertCommitted();
- session.assertNotRolledBack();
- }
-
- private String expectRecordContentIdentity() {
- return """
- {"record":"1"}
- {"record":"1b"}
- {"record":"no-date"}
- {"record":"2"}""";
- }
-
- private String expectRecordContentWrapper() {
- return """
-
{"metadata":{"stream":"kinesis-test","shardId":"another-shard","sequenceNumber":"1","subSequenceNumber":0,\
-
"shardedSequenceNumber":"100000000000000000000","partitionKey":"partition-1","approximateArrival":1609459140000},"value":{"record":"1"}}
-
{"metadata":{"stream":"kinesis-test","shardId":"another-shard","sequenceNumber":"1","subSequenceNumber":0,\
-
"shardedSequenceNumber":"100000000000000000000","partitionKey":"partition-1","approximateArrival":1609459140000},"value":{"record":"1b"}}
-
{"metadata":{"stream":"kinesis-test","shardId":"another-shard","sequenceNumber":"no-date","subSequenceNumber":0,\
-
"shardedSequenceNumber":"no-date00000000000000000000","partitionKey":"partition-no-date","approximateArrival":null},"value":{"record":"no-date"}}
-
{"metadata":{"stream":"kinesis-test","shardId":"another-shard","sequenceNumber":"2","subSequenceNumber":0,\
-
"shardedSequenceNumber":"200000000000000000000","partitionKey":"partition-2","approximateArrival":1609459200000},"value":{"record":"2"}}""";
- }
-
- @ParameterizedTest
- @EnumSource(SchemaDifferenceHandlingStrategy.class)
- public void testProcessPoisonPillRecordButNoRawOutputWithCheckpoint(final
SchemaDifferenceHandlingStrategy strategy) throws ShutdownException,
InvalidStateException {
- fixture = defaultFixtureWithStrategy(strategy);
- final ProcessRecordsInput processRecordsInput =
ProcessRecordsInput.builder()
- .records(Arrays.asList(
-
KinesisClientRecord.builder().approximateArrivalTimestamp(null)
- .partitionKey("partition-1")
- .sequenceNumber("1")
-
.data(ByteBuffer.wrap("{\"record\":\"1\"}".getBytes(StandardCharsets.UTF_8)))
- .build(),
- kinesisRecord,
-
KinesisClientRecord.builder().approximateArrivalTimestamp(null)
- .partitionKey("partition-3")
- .sequenceNumber("3")
-
.data(ByteBuffer.wrap("{\"record\":\"3\"}".getBytes(StandardCharsets.UTF_8)))
- .build()
- ))
- .checkpointer(checkpointer)
- .cacheEntryTime(Instant.now().minus(1, ChronoUnit.MINUTES))
- .cacheExitTime(Instant.now().minus(1, ChronoUnit.SECONDS))
- .millisBehindLatest(100L)
- .build();
-
- when(kinesisRecord.data()).thenThrow(new
IllegalStateException("illegal state"));
- when(kinesisRecord.toString()).thenReturn("poison-pill");
-
- fixture.setKinesisShardId("test-shard");
-
- when(processSessionFactory.createSession()).thenReturn(session);
- fixture.processRecords(processRecordsInput);
- verify(processSessionFactory, times(1)).createSession();
-
- // check non-poison pill records are output successfully
- session.assertTransferCount(ConsumeKinesisStream.REL_SUCCESS, 1);
- final List<MockFlowFile> flowFiles =
session.getFlowFilesForRelationship(ConsumeKinesisStream.REL_SUCCESS);
- // 2 successful records in single output file, attributes equating to
that of the last successful record
- assertFlowFile(flowFiles.getFirst(), null, "partition-3", "3",
"test-shard", "{\"record\":\"1\"}\n" +
- "{\"record\":\"3\"}", 2);
-
- // check no poison-pill output (as the raw data could not be retrieved)
- session.assertTransferCount(ConsumeKinesisStream.REL_PARSE_FAILURE, 0);
-
- // check the "poison pill" record was retried a 2nd time
- assertNull(verify(kinesisRecord, times(2)).data());
- verify(checkpointer, times(1)).checkpoint();
-
- assertFalse(runner.getLogger().getErrorMessages().isEmpty());
-
- session.assertCommitted();
- session.assertNotRolledBack();
- }
-
- private static Stream<Arguments.ArgumentSet> unparsableRecordsLists() {
- final KinesisClientRecord unparsableRecordMock =
mock(KinesisClientRecord.class);
- final KinesisClientRecord parsableRecord1 =
KinesisClientRecord.builder().approximateArrivalTimestamp(null)
- .partitionKey("partition-1")
- .sequenceNumber("1")
-
.data(ByteBuffer.wrap("{\"record\":\"1\"}".getBytes(StandardCharsets.UTF_8)))
- .build();
- final KinesisClientRecord parsableRecord3 =
KinesisClientRecord.builder().approximateArrivalTimestamp(null)
- .partitionKey("partition-3")
- .sequenceNumber("3")
-
.data(ByteBuffer.wrap("{\"record\":\"3\"}".getBytes(StandardCharsets.UTF_8)))
- .build();
- final List<List<KinesisClientRecord>> recordLists = List.of(
- Arrays.asList(
- unparsableRecordMock,
- parsableRecord1,
- parsableRecord3
- ),
- Arrays.asList(
- parsableRecord1,
- unparsableRecordMock,
- parsableRecord3
- ),
- Arrays.asList(
- parsableRecord1,
- parsableRecord3,
- unparsableRecordMock
- )
- );
- return Arrays.stream(SchemaDifferenceHandlingStrategy.values())
- .flatMap(strategy -> recordLists.stream().map(recordList ->
Pair.of(strategy, recordList)))
- .map(pair -> {
- final SchemaDifferenceHandlingStrategy strategy =
pair.getLeft();
- final List<KinesisClientRecord> inputRecords =
pair.getRight();
- String position = "Middle";
- if (inputRecords.getFirst() == unparsableRecordMock) {
- position = "Beginning";
- }
- if (inputRecords.getLast() == unparsableRecordMock) {
- position = "End";
- }
- return Arguments.argumentSet(
- String.format("Strategy=%s, Unparsable Record
Position=%s", strategy, position),
- inputRecords,
- unparsableRecordMock,
- strategy
- );
- })
- .peek(it -> {
- parsableRecord1.data().rewind();
- parsableRecord3.data().rewind();
- reset(unparsableRecordMock);
- });
- }
-
- @ParameterizedTest
- @MethodSource("unparsableRecordsLists")
- void testProcessUnparsableRecordWithRawOutputWithCheckpoint(
- final List<KinesisClientRecord> inputRecords,
- final KinesisClientRecord kinesisRecord,
- final SchemaDifferenceHandlingStrategy strategy) throws
ShutdownException, InvalidStateException {
- fixture = defaultFixtureWithStrategy(strategy);
- final ProcessRecordsInput processRecordsInput =
ProcessRecordsInput.builder()
- .records(inputRecords)
- .checkpointer(checkpointer)
- .cacheEntryTime(Instant.now().minus(1, ChronoUnit.MINUTES))
- .cacheExitTime(Instant.now().minus(1, ChronoUnit.SECONDS))
- .millisBehindLatest(100L)
- .build();
-
-
when(kinesisRecord.data()).thenReturn(ByteBuffer.wrap("invalid-json".getBytes(StandardCharsets.UTF_8)));
- when(kinesisRecord.partitionKey()).thenReturn("unparsable-partition");
- when(kinesisRecord.sequenceNumber()).thenReturn("unparsable-sequence");
- when(kinesisRecord.approximateArrivalTimestamp()).thenReturn(null);
-
- fixture.setKinesisShardId("test-shard");
-
- when(processSessionFactory.createSession()).thenReturn(session);
- fixture.processRecords(processRecordsInput);
- verify(processSessionFactory, times(1)).createSession();
-
- // check non-poison pill records are output successfully
- session.assertTransferCount(ConsumeKinesisStream.REL_SUCCESS, 1);
- final List<MockFlowFile> flowFiles =
session.getFlowFilesForRelationship(ConsumeKinesisStream.REL_SUCCESS);
- // 2 successful records in single output file, attributes equating to
that of the last successful record
- assertFlowFile(flowFiles.getFirst(), null, "partition-3", "3",
"test-shard", "{\"record\":\"1\"}\n" +
- "{\"record\":\"3\"}", 2);
-
- // check poison-pill output (as the raw data could not be retrieved)
- session.assertTransferCount(ConsumeKinesisStream.REL_PARSE_FAILURE, 1);
- final List<MockFlowFile> failureFlowFiles =
session.getFlowFilesForRelationship(ConsumeKinesisStream.REL_PARSE_FAILURE);
- assertFlowFile(failureFlowFiles.getFirst(), null,
"unparsable-partition", "unparsable-sequence", "test-shard", "invalid-json", 0);
-
failureFlowFiles.getFirst().assertAttributeExists("record.error.message");
-
- // check the invalid json record was *not* retried a 2nd time
- assertNull(verify(kinesisRecord, times(1)).partitionKey());
- assertNull(verify(kinesisRecord, times(1)).sequenceNumber());
- assertNull(verify(kinesisRecord,
times(1)).approximateArrivalTimestamp());
- assertNull(verify(kinesisRecord, times(1)).data());
- verify(checkpointer, times(1)).checkpoint();
-
- assertEquals(1, runner.getLogger().getErrorMessages().size());
-
- session.assertCommitted();
- session.assertNotRolledBack();
- }
-
- @ParameterizedTest
- @EnumSource(SchemaDifferenceHandlingStrategy.class)
- void testProcessMultipleRecordInSingleKinesisRecord(final
SchemaDifferenceHandlingStrategy strategy) throws ShutdownException,
InvalidStateException {
- fixture = defaultFixtureWithStrategy(strategy);
-
testFlowFileContents(List.of("[{\"record\":1},{\"record\":2},{\"record\":3}]"),
List.of(EmittedFlowFile.of("{\"record\":1}\n{\"record\":2}\n{\"record\":3}",
3)));
- }
-
- @Nested
- class StateHandlerStrategyTest {
-
- private static List<String> inputJsonRecords() {
- return List.of(
- "{\"colA\":1}", // inferred to ["colA":int]
- "{\"colA\":%d}".formatted(Long.MAX_VALUE), // inferred to
["colA":long]
- "{\"colA\":3}"
- );
- }
-
- @Test
- void testProcessIncompatibleSchemaKinesisRecordsStrategyGrouping()
throws ShutdownException, InvalidStateException {
- fixture =
defaultFixtureWithStrategy(SchemaDifferenceHandlingStrategy.GROUP_RECORDS);
- final List<String> inputJsonRecords = inputJsonRecords();
- testFlowFileContents(
- inputJsonRecords,
- List.of(
- EmittedFlowFile.of(String.join("\n",
inputJsonRecords.get(0), inputJsonRecords.get(2)), 2),
- EmittedFlowFile.of(inputJsonRecords.get(1), 1)
- )
- );
- }
-
- @Test
- void testProcessIncompatibleSchemaKinesisRecordsStrategyRolling()
throws ShutdownException, InvalidStateException {
- fixture =
defaultFixtureWithStrategy(SchemaDifferenceHandlingStrategy.CREATE_FLOW_FILE);
- final List<String> inputJsonRecords = inputJsonRecords();
- testFlowFileContents(
- inputJsonRecords,
- List.of(
- EmittedFlowFile.of(inputJsonRecords.get(0), 1),
- EmittedFlowFile.of(inputJsonRecords.get(1), 1),
- EmittedFlowFile.of(inputJsonRecords.get(2), 1)
- )
- );
- }
- }
-
- void testFlowFileContents(final List<String> kinesisRecordJsonContents,
final List<EmittedFlowFile> emittedFlowFiles) throws ShutdownException,
InvalidStateException {
- final List<KinesisClientRecord> kinesisRecords =
kinesisRecordJsonContents.stream()
- .map(json -> KinesisClientRecord.builder()
- .approximateArrivalTimestamp(null)
- .partitionKey("partition-1")
- .sequenceNumber("1")
-
.data(ByteBuffer.wrap(json.getBytes(StandardCharsets.UTF_8)))
- .build())
- .toList();
- final ProcessRecordsInput processRecordsInput =
ProcessRecordsInput.builder()
- .records(kinesisRecords)
- .checkpointer(checkpointer)
- .cacheEntryTime(Instant.now().minus(1, ChronoUnit.MINUTES))
- .cacheExitTime(Instant.now().minus(1, ChronoUnit.SECONDS))
- .millisBehindLatest(100L)
- .build();
-
-
when(kinesisRecord.data()).thenReturn(ByteBuffer.wrap("invalid-json".getBytes(StandardCharsets.UTF_8)));
- when(kinesisRecord.partitionKey()).thenReturn("unparsable-partition");
- when(kinesisRecord.sequenceNumber()).thenReturn("unparsable-sequence");
- when(kinesisRecord.approximateArrivalTimestamp()).thenReturn(null);
-
- fixture.setKinesisShardId("test-shard");
-
- when(processSessionFactory.createSession()).thenReturn(session);
- fixture.processRecords(processRecordsInput);
- verify(processSessionFactory, times(1)).createSession();
-
- // check non-poison pill records are output successfully
- session.assertTransferCount(ConsumeKinesisStream.REL_SUCCESS,
emittedFlowFiles.size());
- final List<MockFlowFile> flowFiles =
session.getFlowFilesForRelationship(ConsumeKinesisStream.REL_SUCCESS);
- for (int i = 0; i < emittedFlowFiles.size(); i++) {
- final EmittedFlowFile emittedFlowFile = emittedFlowFiles.get(i);
- assertFlowFile(flowFiles.get(i), null, "partition-1", "1",
"test-shard", emittedFlowFile.content, emittedFlowFile.recordCount);
- }
-
- verify(checkpointer, times(1)).checkpoint();
-
- session.assertCommitted();
- session.assertNotRolledBack();
- }
-
- private record EmittedFlowFile(String content, int recordCount) {
- public static EmittedFlowFile of(final String content, final int
recordCount) {
- return new EmittedFlowFile(content, recordCount);
- }
- }
-
- private void assertFlowFile(final MockFlowFile flowFile, final Date
approxTimestamp, final String partitionKey,
- final String sequenceNumber, final String
shard, final String content, final int recordCount) {
- if (approxTimestamp != null) {
-
flowFile.assertAttributeEquals(AbstractKinesisRecordProcessor.AWS_KINESIS_APPROXIMATE_ARRIVAL_TIMESTAMP,
-
DATE_TIME_FORMATTER.format(ZonedDateTime.ofInstant(approxTimestamp.toInstant(),
ZoneId.systemDefault())));
- } else {
-
flowFile.assertAttributeNotExists(AbstractKinesisRecordProcessor.AWS_KINESIS_APPROXIMATE_ARRIVAL_TIMESTAMP);
- }
-
flowFile.assertAttributeEquals(AbstractKinesisRecordProcessor.AWS_KINESIS_PARTITION_KEY,
partitionKey);
-
flowFile.assertAttributeEquals(AbstractKinesisRecordProcessor.AWS_KINESIS_SEQUENCE_NUMBER,
sequenceNumber);
-
flowFile.assertAttributeEquals(AbstractKinesisRecordProcessor.AWS_KINESIS_SHARD_ID,
shard);
-
- if (recordCount > 0) {
- flowFile.assertAttributeEquals("record.count",
String.valueOf(recordCount));
- flowFile.assertAttributeEquals(CoreAttributes.MIME_TYPE.key(),
"application/json");
- } else {
- flowFile.assertAttributeNotExists("record.count");
- flowFile.assertAttributeNotExists(CoreAttributes.MIME_TYPE.key());
- }
-
- flowFile.assertContentEquals(content);
- }
-}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/conf/logback.xml
b/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/conf/logback.xml
index d42d71bcac4..1ed2bef1b69 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/conf/logback.xml
+++
b/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/conf/logback.xml
@@ -210,9 +210,6 @@
<!-- Suppress non-error messages due to Jetty AnnotationParser emitting a
large amount of WARNS. Issue described in NIFI-5479. -->
<logger name="org.eclipse.jetty.annotations.AnnotationParser"
level="ERROR"/>
- <!-- Suppress non-error messages from AWS KCL which was emitting large
amounts of INFO logs by default -->
- <logger name="software.amazon.awssdk.kinesis" level="WARN" />
-
<!-- Suppress non-error messages from Apache Atlas which was emitting
large amounts of INFO logs by default -->
<logger name="org.apache.atlas" level="WARN"/>
diff --git
a/nifi-stateless/nifi-stateless-resources/src/main/resources/conf/stateless-logback.xml
b/nifi-stateless/nifi-stateless-resources/src/main/resources/conf/stateless-logback.xml
index ad915a96d3d..6ea9a5c28f4 100644
---
a/nifi-stateless/nifi-stateless-resources/src/main/resources/conf/stateless-logback.xml
+++
b/nifi-stateless/nifi-stateless-resources/src/main/resources/conf/stateless-logback.xml
@@ -60,9 +60,6 @@
<logger name="org.apache.calcite.runtime.CalciteException" level="OFF" />
- <!-- Suppress non-error messages from AWS KCL which was emitting large
amounts of INFO logs by default -->
- <logger name="software.amazon.awssdk.kinesis" level="WARN" />
-
<root level="INFO">
<appender-ref ref="CONSOLE" />
<appender-ref ref="APP_FILE" />
diff --git
a/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/clustered/node1/logback.xml
b/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/clustered/node1/logback.xml
index 263aaca4843..ee1f1d7802b 100644
---
a/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/clustered/node1/logback.xml
+++
b/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/clustered/node1/logback.xml
@@ -136,9 +136,6 @@
<!-- Suppress non-error messages from SMBJ which was emitting large
amounts of INFO logs by default -->
<logger name="com.hierynomus.smbj" level="WARN" />
- <!-- Suppress non-error messages from AWS KCL which was emitting large
amounts of INFO logs by default -->
- <logger name="software.amazon.awssdk.kinesis" level="WARN" />
-
<!-- Suppress non-error messages from Apache Atlas which was emitting
large amounts of INFO logs by default -->
<logger name="org.apache.atlas" level="WARN"/>
diff --git
a/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/clustered/node2/logback.xml
b/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/clustered/node2/logback.xml
index 263aaca4843..ee1f1d7802b 100644
---
a/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/clustered/node2/logback.xml
+++
b/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/clustered/node2/logback.xml
@@ -136,9 +136,6 @@
<!-- Suppress non-error messages from SMBJ which was emitting large
amounts of INFO logs by default -->
<logger name="com.hierynomus.smbj" level="WARN" />
- <!-- Suppress non-error messages from AWS KCL which was emitting large
amounts of INFO logs by default -->
- <logger name="software.amazon.awssdk.kinesis" level="WARN" />
-
<!-- Suppress non-error messages from Apache Atlas which was emitting
large amounts of INFO logs by default -->
<logger name="org.apache.atlas" level="WARN"/>
diff --git
a/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/default/logback.xml
b/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/default/logback.xml
index d965f00d60e..dd09821a140 100644
---
a/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/default/logback.xml
+++
b/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/default/logback.xml
@@ -136,9 +136,6 @@
<!-- Suppress non-error messages from SMBJ which was emitting large
amounts of INFO logs by default -->
<logger name="com.hierynomus.smbj" level="WARN" />
- <!-- Suppress non-error messages from AWS KCL which was emitting large
amounts of INFO logs by default -->
- <logger name="software.amazon.awssdk.kinesis" level="WARN" />
-
<!-- Suppress non-error messages from Apache Atlas which was emitting
large amounts of INFO logs by default -->
<logger name="org.apache.atlas" level="WARN"/>
diff --git
a/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/pythonic/logback.xml
b/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/pythonic/logback.xml
index 637c570239b..7fc296f8146 100644
---
a/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/pythonic/logback.xml
+++
b/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/pythonic/logback.xml
@@ -142,9 +142,6 @@
<!-- Suppress non-error messages from SMBJ which was emitting large
amounts of INFO logs by default -->
<logger name="com.hierynomus.smbj" level="WARN" />
- <!-- Suppress non-error messages from AWS KCL which was emitting large
amounts of INFO logs by default -->
- <logger name="com.amazonaws.services.kinesis" level="WARN" />
-
<!-- Suppress non-error messages from Apache Atlas which was emitting
large amounts of INFO logs by default -->
<logger name="org.apache.atlas" level="WARN"/>
