Hyeongrak JI created NIFI-16485:
-----------------------------------
Summary: Add optional Max Poll Records override to ConsumeKafka
Key: NIFI-16485
URL: https://issues.apache.org/jira/browse/NIFI-16485
Project: Apache NiFi
Issue Type: Improvement
Components: Extensions
Reporter: Hyeongrak JI
The {{ConsumeKafka}} Processor should support an optional {{Max Poll Records}}
property that overrides the {{max.poll.records}} value of
{{Kafka3ConnectionService}} for that Processor. It is a tuning value whose
right setting depends on the size of the records being read. Record sizes
differ between topics, the topic is set on the Processor, and one Connection
Service can serve Processors that read different topics. Today, using a
different value for one Processor requires a separate Connection Service that
repeats the connection and security settings only to change this one consumer
setting.
{{max.poll.records}} is the maximum number of records a single poll returns.
With small records, a high value lets each poll return more records; with large
records, a lower value limits how much data each poll hands to the Processor.
ConsumeKafka also checks {{Max Uncommitted Size}} only after processing all
records from a poll, so a single invocation can exceed Max Uncommitted Size by
up to {{max.poll.records}} records. Max Uncommitted Size is set on the
Processor, but the value that limits this overrun is set on the Connection
Service.
In NiFi 1.x, Max Poll Records ({{max.poll.records}}, default 10000) was a
property of {{ConsumeKafka_2_0}} and {{ConsumeKafka_2_6}}. In a production NiFi
1.19 cluster we run, 44 of 446 running ConsumeKafka Processors use a value
other than 10000. 40 of them are in 16 sets of Processors that share the same
brokers, security settings, and credentials but use different values. Migrating
these flows one-to-one to 2.x requires additional Connection Services for this
value alone.
The override follows how {{Kafka3ConnectionService}} already combines service
and Processor settings. ConsumeKafka supplies {{group.id}} and
{{auto.offset.reset}} through {{PollingContext}}, and PublishKafka supplies
{{acks}}, {{compression.type}}, {{max.request.size}}, and {{partitioner.class}}
through {{ProducerConfiguration}}, which {{getProducerService}} applies over
the service properties. Connecting and authenticating to the cluster stay with
the Connection Service.
h3. Proposed change
* Add an optional {{Max Poll Records}} property to {{ConsumeKafka}} with no
default value. When it is not set, the Connection Service value applies and
existing flows behave as before.
* Carry the value in {{PollingContext}}, alongside the Group ID and Auto Offset
Reset.
* Apply it in {{Kafka3ConnectionService.getConsumerService}} after the service
consumer properties.
I can submit a pull request with unit tests.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
