You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by pv...@apache.org on 2024/05/23 18:05:36 UTC
(nifi) branch main updated: NIFI-13283 - Fix exception thrown in PutKinesisFirehose processor by ensuring value exists before referencing
This is an automated email from the ASF dual-hosted git repository.
pvillard 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 99253b9673 NIFI-13283 - Fix exception thrown in PutKinesisFirehose processor by ensuring value exists before referencing
99253b9673 is described below
commit 99253b9673ac7aab19e6612599a0e687d8eb438f
Author: Evan Shelton <ev...@gmail.com>
AuthorDate: Wed May 22 14:08:04 2024 -0400
NIFI-13283 - Fix exception thrown in PutKinesisFirehose processor by ensuring value exists before referencing
Signed-off-by: Pierre Villard <pi...@gmail.com>
This closes #8867.
---
.../apache/nifi/processors/aws/kinesis/firehose/PutKinesisFirehose.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/java/org/apache/nifi/processors/aws/kinesis/firehose/PutKinesisFirehose.java b/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/java/org/apache/nifi/processors/aws/kinesis/firehose/PutKinesisFirehose.java
index 46ba510b10..3f810070b2 100644
--- a/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/java/org/apache/nifi/processors/aws/kinesis/firehose/PutKinesisFirehose.java
+++ b/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/java/org/apache/nifi/processors/aws/kinesis/firehose/PutKinesisFirehose.java
@@ -130,9 +130,9 @@ public class PutKinesisFirehose extends AbstractAwsSyncProcessor<FirehoseClient,
for (final FlowFile flowFile : flowFiles) {
final String firehoseStreamName = context.getProperty(KINESIS_FIREHOSE_DELIVERY_STREAM_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ recordHash.computeIfAbsent(firehoseStreamName, k -> new ArrayList<>());
session.read(flowFile, in -> recordHash.get(firehoseStreamName).add(Record.builder().data(SdkBytes.fromInputStream(in)).build()));
- recordHash.computeIfAbsent(firehoseStreamName, k -> new ArrayList<>());
final List<FlowFile> flowFilesForStream = hashFlowFiles.computeIfAbsent(firehoseStreamName, k -> new ArrayList<>());
flowFilesForStream.add(flowFile);
}