[
https://issues.apache.org/jira/browse/NIFI-16187?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18106260#comment-18106260
]
ASF subversion and git services commented on NIFI-16187:
--------------------------------------------------------
Commit 1a22684df8d7ea3aa8bb0952c001894b16c85d0a in nifi's branch
refs/heads/main from Alaksiej Ščarbaty
[ https://gitbox.apache.org/repos/asf?p=nifi.git;h=1a22684df8d ]
NIFI-16187 FlowFile grouping strategy in ConsumeKafka (#11537)
Extract FlowFile grouping into RecordGroupingStrategy, then add
MergeSchemaGrouping
and a Schema Conflict Resolution property so RECORD consumption can batch
different
schemas into one FlowFile per topic/partition group while preserving Create New
FlowFile as the default.
Make DataTypeUtils.merge nullify fields present in only one schema so a merged
schema can describe records from either input without failing strict writers.
> Write messages with different schemas into same FlowFiles in ConsumeKafka
> -------------------------------------------------------------------------
>
> Key: NIFI-16187
> URL: https://issues.apache.org/jira/browse/NIFI-16187
> Project: Apache NiFi
> Issue Type: Improvement
> Components: Extensions
> Affects Versions: 2.11.0
> Reporter: Alaksiej Ščarbaty
> Assignee: Alaksiej Ščarbaty
> Priority: Major
> Time Spent: 1h 20m
> Remaining Estimate: 0h
>
> h2. Problem
> When _ConsumeKafka_ is configured with _RECORD_ processing strategy, it uses
> nifi record reader and writer for serde operations. Incoming messages are
> written into FlowFiles in batches, but only if their schemas are identical.
> In case there is a schema mismatch, a new FlowFile is created.
> This works poorly for semi-structured json messages with schema inference, as
> a lot of tiny FlowFiles are created. Very often with a single record per
> FlowFile, which negatively affects the performance of the downstream
> processors.
> h2. Suggestion
> Add a new {{_Schema Conflict Resolution_}} option to the processor.
>
> Its default value is {{{}_Create New FlowFile_{}}}, which will represent the
> current behavior of the processor. Nothing is going to be changes in this
> path - no risk of introducing degradation to existing customers.
> Another value would be {_}Continue with Merged Schema{_}. The processor reads
> a batch of incoming Kafka messages, which are already in memory, into a list.
> # The processor iterates over the records and merges their read schemas into
> a single write schema.
> # The processor creates a single record writer with the merged schema, the
> records are written into a single FlowFile using this schema.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
