This is an automated email from the ASF dual-hosted git repository.
markap14 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 62e4d95a824 NIFI-16554 Expose Kafka fetch size properties in
Kafka3ConnectionService (#11894)
62e4d95a824 is described below
commit 62e4d95a824e87e5a81bf29bc572e8626d9a8519
Author: Wojciech Trefon <[email protected]>
AuthorDate: Fri Nov 6 21:00:25 2026 +0100
NIFI-16554 Expose Kafka fetch size properties in Kafka3ConnectionService
(#11894)
---
.../additionalDetails.md | 12 +++
.../service/aws/AmazonMSKConnectionService.java | 2 +
.../kafka/service/Kafka3ConnectionService.java | 53 +++++++++
.../kafka/service/Kafka3ConnectionServiceTest.java | 119 +++++++++++++++++++++
4 files changed, 186 insertions(+)
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/resources/docs/org.apache.nifi.kafka.processors.ConsumeKafka/additionalDetails.md
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/resources/docs/org.apache.nifi.kafka.processors.ConsumeKafka/additionalDetails.md
index e454c53f027..fb5f455cd16 100644
---
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/resources/docs/org.apache.nifi.kafka.processors.ConsumeKafka/additionalDetails.md
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/resources/docs/org.apache.nifi.kafka.processors.ConsumeKafka/additionalDetails.md
@@ -175,3 +175,15 @@ Here is an example of FlowFile content that is emitted by
JsonRecordSetWriter wh
}
]
```
+
+### Memory Usage with Continue with Merged Schema
+
+When the Schema Conflict Resolution property is set to 'Continue with Merged
Schema', the records from an entire poll are grouped by topic and partition and
held in memory until they are written to FlowFiles.
+Each concurrent task holds its own poll, and when the consumer falls behind,
every poll is full.
+
+Memory usage can be limited by lowering the following properties of the Kafka
Connection Service:
+
+- 'Max Partition Fetch Size' limits how much data the broker returns for each
partition in a single fetch.
+- 'Max Poll Records' limits how many records a single poll returns.
+
+Lower values reduce throughput: smaller fetches mean more requests to the
broker, and fewer records per poll mean smaller FlowFiles.
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-aws/src/main/java/org/apache/nifi/kafka/service/aws/AmazonMSKConnectionService.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-aws/src/main/java/org/apache/nifi/kafka/service/aws/AmazonMSKConnectionService.java
index 1cea4b71e85..faed6236d1b 100644
---
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-aws/src/main/java/org/apache/nifi/kafka/service/aws/AmazonMSKConnectionService.java
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-aws/src/main/java/org/apache/nifi/kafka/service/aws/AmazonMSKConnectionService.java
@@ -357,6 +357,8 @@ public class AmazonMSKConnectionService extends
Kafka3ConnectionService {
@Override
public void migrateProperties(final PropertyConfiguration config) {
+ super.migrateProperties(config);
+
// For backward compatibility: if an AWS Profile Name was configured
previously,
// set AWS Role Source to SPECIFIED_PROFILE
if (config.isPropertySet(AWS_PROFILE_NAME)) {
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/main/java/org/apache/nifi/kafka/service/Kafka3ConnectionService.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/main/java/org/apache/nifi/kafka/service/Kafka3ConnectionService.java
index b972e076f13..949beffa10c 100644
---
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/main/java/org/apache/nifi/kafka/service/Kafka3ConnectionService.java
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/main/java/org/apache/nifi/kafka/service/Kafka3ConnectionService.java
@@ -67,7 +67,9 @@ import
org.apache.nifi.kafka.shared.property.provider.StandardKafkaPropertyProvi
import org.apache.nifi.kafka.shared.transaction.TransactionIdSupplier;
import org.apache.nifi.kafka.shared.validation.DynamicPropertyValidator;
import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.migration.PropertyConfiguration;
import org.apache.nifi.oauth2.OAuth2AccessTokenProvider;
+import org.apache.nifi.processor.DataUnit;
import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.ssl.SSLContextProvider;
@@ -128,6 +130,32 @@ public class Kafka3ConnectionService extends
AbstractControllerService implement
.addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
.build();
+ public static final PropertyDescriptor MAX_PARTITION_FETCH_SIZE = new
PropertyDescriptor.Builder()
+ .name("Max Partition Fetch Size")
+ .description("""
+ Maximum amount of data that the broker returns for each
partition in a single fetch.
+ When a consumer is assigned only a few partitions, this
limit determines how much data each poll returns and how much each consumer
holds in memory.
+ A record batch larger than the limit is still returned so
that the consumer can make progress.
+ Corresponds to Kafka max.partition.fetch.bytes property.
+ """)
+ .required(true)
+ .defaultValue("1 MB")
+ .addValidator(StandardValidators.createDataSizeBoundsValidator(0,
Integer.MAX_VALUE))
+ .build();
+
+ public static final PropertyDescriptor MAX_FETCH_SIZE = new
PropertyDescriptor.Builder()
+ .name("Max Fetch Size")
+ .description("""
+ Maximum amount of data that the broker returns for a
single fetch request across all partitions.
+ Applies only when the per-partition limits of the
partitions fetched from one broker add up to more than this value.
+ A record batch larger than the limit is still returned so
that the consumer can make progress.
+ Corresponds to Kafka fetch.max.bytes property.
+ """)
+ .required(true)
+ .defaultValue("50 MB")
+ .addValidator(StandardValidators.createDataSizeBoundsValidator(0,
Integer.MAX_VALUE))
+ .build();
+
public static final PropertyDescriptor CLIENT_TIMEOUT = new
PropertyDescriptor.Builder()
.name("default.api.timeout.ms")
.displayName("Client Timeout")
@@ -176,6 +204,8 @@ public class Kafka3ConnectionService extends
AbstractControllerService implement
SSL_CONTEXT_SERVICE,
TRANSACTION_ISOLATION_LEVEL,
MAX_POLL_RECORDS,
+ MAX_PARTITION_FETCH_SIZE,
+ MAX_FETCH_SIZE,
CLIENT_TIMEOUT,
METADATA_WAIT_TIME,
ACK_WAIT_TIME
@@ -183,6 +213,8 @@ public class Kafka3ConnectionService extends
AbstractControllerService implement
private static final KafkaConnectionVerifier kafkaConnectionVerifier = new
KafkaConnectionVerifier();
+ private static final Pattern BYTES_PATTERN = Pattern.compile("\\d+");
+
/**
* Bound applied to the Admin client used only for backlog determination,
in place of the
* potentially much longer Client Timeout configured for regular
produce/consume operations.
@@ -257,6 +289,21 @@ public class Kafka3ConnectionService extends
AbstractControllerService implement
.build();
}
+ @Override
+ public void migrateProperties(final PropertyConfiguration config) {
+ migrateBytesProperty(config,
ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, MAX_PARTITION_FETCH_SIZE);
+ migrateBytesProperty(config, ConsumerConfig.FETCH_MAX_BYTES_CONFIG,
MAX_FETCH_SIZE);
+ }
+
+ private void migrateBytesProperty(final PropertyConfiguration config,
final String bytesPropertyName, final PropertyDescriptor dataSizeProperty) {
+ if (config.renameProperty(bytesPropertyName,
dataSizeProperty.getName())) {
+ config.getRawPropertyValue(dataSizeProperty)
+ .map(String::trim)
+ .filter(value -> BYTES_PATTERN.matcher(value).matches())
+ .ifPresent(value -> config.setProperty(dataSizeProperty,
value + " B"));
+ }
+ }
+
@Override
public KafkaConsumerService getConsumerService(final PollingContext
pollingContext) {
Objects.requireNonNull(pollingContext, "Polling Context required");
@@ -490,6 +537,12 @@ public class Kafka3ConnectionService extends
AbstractControllerService implement
final IsolationLevel isolationLevel =
propertyContext.getProperty(TRANSACTION_ISOLATION_LEVEL).asAllowableValue(IsolationLevel.class);
properties.put(TRANSACTION_ISOLATION_LEVEL.getName(),
isolationLevel.getValue());
+ final int maxPartitionFetchSize =
propertyContext.getProperty(MAX_PARTITION_FETCH_SIZE).asDataSize(DataUnit.B).intValue();
+ properties.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG,
maxPartitionFetchSize);
+
+ final int maxFetchSize =
propertyContext.getProperty(MAX_FETCH_SIZE).asDataSize(DataUnit.B).intValue();
+ properties.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, maxFetchSize);
+
return properties;
}
diff --git
a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/test/java/org/apache/nifi/kafka/service/Kafka3ConnectionServiceTest.java
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/test/java/org/apache/nifi/kafka/service/Kafka3ConnectionServiceTest.java
new file mode 100644
index 00000000000..84e591cd81d
--- /dev/null
+++
b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/test/java/org/apache/nifi/kafka/service/Kafka3ConnectionServiceTest.java
@@ -0,0 +1,119 @@
+/*
+ * 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.kafka.service;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.reporting.InitializationException;
+import org.apache.nifi.util.MockConfigurationContext;
+import org.apache.nifi.util.MockProcessContext;
+import org.apache.nifi.util.MockPropertyConfiguration;
+import org.apache.nifi.util.NoOpProcessor;
+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 java.util.Collections;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+class Kafka3ConnectionServiceTest {
+
+ private static final String SERVICE_ID =
Kafka3ConnectionService.class.getSimpleName();
+
+ private static final String BOOTSTRAP_SERVERS = "localhost:9092";
+
+ private static final String MAX_PARTITION_FETCH_SIZE = "256 KB";
+
+ private static final int MAX_PARTITION_FETCH_SIZE_BYTES = 262144;
+
+ private static final String MAX_FETCH_SIZE = "10 MB";
+
+ private static final int MAX_FETCH_SIZE_BYTES = 10485760;
+
+ private TestRunner runner;
+
+ private Kafka3ConnectionService service;
+
+ @BeforeEach
+ void setRunner() throws InitializationException {
+ runner = TestRunners.newTestRunner(NoOpProcessor.class);
+ service = new Kafka3ConnectionService();
+ runner.addControllerService(SERVICE_ID, service);
+ runner.setProperty(service, Kafka3ConnectionService.BOOTSTRAP_SERVERS,
BOOTSTRAP_SERVERS);
+ }
+
+ @Test
+ void testFetchSizeInConsumerProperties() {
+ runner.assertValid(service);
+
+ final Properties defaultConsumerProperties = getConsumerProperties();
+
+ assertEquals(ConsumerConfig.DEFAULT_MAX_PARTITION_FETCH_BYTES,
defaultConsumerProperties.get(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG));
+ assertEquals(ConsumerConfig.DEFAULT_FETCH_MAX_BYTES,
defaultConsumerProperties.get(ConsumerConfig.FETCH_MAX_BYTES_CONFIG));
+
+ runner.setProperty(service,
Kafka3ConnectionService.MAX_PARTITION_FETCH_SIZE, MAX_PARTITION_FETCH_SIZE);
+ runner.setProperty(service, Kafka3ConnectionService.MAX_FETCH_SIZE,
MAX_FETCH_SIZE);
+ runner.assertValid(service);
+
+ final Properties consumerProperties = getConsumerProperties();
+
+ assertEquals(MAX_PARTITION_FETCH_SIZE_BYTES,
consumerProperties.get(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG));
+ assertEquals(MAX_FETCH_SIZE_BYTES,
consumerProperties.get(ConsumerConfig.FETCH_MAX_BYTES_CONFIG));
+ }
+
+ @Test
+ void testMigrateProperties() {
+ final Map<String, String> propertyValues = Map.of(
+ ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG,
Integer.toString(MAX_PARTITION_FETCH_SIZE_BYTES),
+ ConsumerConfig.FETCH_MAX_BYTES_CONFIG,
Integer.toString(MAX_FETCH_SIZE_BYTES)
+ );
+ final MockPropertyConfiguration configuration = new
MockPropertyConfiguration(propertyValues);
+
+ service.migrateProperties(configuration);
+
+ final Map<String, String> expectedProperties = Map.of(
+ Kafka3ConnectionService.MAX_PARTITION_FETCH_SIZE.getName(),
MAX_PARTITION_FETCH_SIZE_BYTES + " B",
+ Kafka3ConnectionService.MAX_FETCH_SIZE.getName(),
MAX_FETCH_SIZE_BYTES + " B"
+ );
+ assertEquals(expectedProperties, configuration.getRawProperties());
+ }
+
+ @Test
+ void testMigratePropertiesExpressionLanguageNotConverted() {
+ final String expression = "${MAX_PARTITION_FETCH_BYTES}";
+ final Map<String, String> propertyValues =
Map.of(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, expression);
+ final MockPropertyConfiguration configuration = new
MockPropertyConfiguration(propertyValues);
+
+ service.migrateProperties(configuration);
+
+ final Map<String, String> expectedProperties =
Map.of(Kafka3ConnectionService.MAX_PARTITION_FETCH_SIZE.getName(), expression);
+ assertEquals(expectedProperties, configuration.getRawProperties());
+ }
+
+ private Properties getConsumerProperties() {
+ final MockProcessContext processContext = (MockProcessContext)
runner.getProcessContext();
+ final Map<PropertyDescriptor, String> properties =
processContext.getConfiguration(SERVICE_ID).getProperties();
+ final MockConfigurationContext configurationContext = new
MockConfigurationContext(service, properties, processContext,
Collections.emptyMap());
+
+ final Properties clientProperties =
service.getClientProperties(configurationContext);
+ return service.getConsumerProperties(configurationContext,
clientProperties);
+ }
+}
