markap14 commented on code in PR #11894:
URL: https://github.com/apache/nifi/pull/11894#discussion_r4493382310
##########
nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/main/java/org/apache/nifi/kafka/service/Kafka3ConnectionService.java:
##########
@@ -128,6 +128,36 @@ public class Kafka3ConnectionService extends
AbstractControllerService implement
.addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
.build();
+ public static final PropertyDescriptor MAX_PARTITION_FETCH_BYTES = new
PropertyDescriptor.Builder()
+ .name("max.partition.fetch.bytes")
+ .displayName("Max Partition Fetch Bytes")
Review Comment:
Please use only `.name(...)` here, without `.displayName(...)`. Splitting
name and display name is an old pattern that we're moving away from, even in
components that still mix the two.
The name should also describe a data size rather than a unit, such as `Max
Partition Fetch Size`. Properties shouldn't mention "bytes" or make the user
enter a value in a fixed unit. They should accept values such as `256 KB` or `1
MB`. Renaming means existing flows need migration; see my comment on the
property list below.
##########
nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/main/java/org/apache/nifi/kafka/service/Kafka3ConnectionService.java:
##########
@@ -128,6 +128,36 @@ public class Kafka3ConnectionService extends
AbstractControllerService implement
.addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
.build();
+ public static final PropertyDescriptor MAX_PARTITION_FETCH_BYTES = new
PropertyDescriptor.Builder()
+ .name("max.partition.fetch.bytes")
+ .displayName("Max Partition Fetch Bytes")
+ .description("""
+ Maximum amount of data in bytes 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(String.valueOf(ConsumerConfig.DEFAULT_MAX_PARTITION_FETCH_BYTES))
+ .addValidator(StandardValidators.NON_NEGATIVE_INTEGER_VALIDATOR)
+ .expressionLanguageSupported(ExpressionLanguageScope.ENVIRONMENT)
Review Comment:
No need for Expression Language support here.
##########
nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/main/java/org/apache/nifi/kafka/service/Kafka3ConnectionService.java:
##########
@@ -128,6 +128,36 @@ public class Kafka3ConnectionService extends
AbstractControllerService implement
.addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
.build();
+ public static final PropertyDescriptor MAX_PARTITION_FETCH_BYTES = new
PropertyDescriptor.Builder()
+ .name("max.partition.fetch.bytes")
+ .displayName("Max Partition Fetch Bytes")
+ .description("""
+ Maximum amount of data in bytes 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(String.valueOf(ConsumerConfig.DEFAULT_MAX_PARTITION_FETCH_BYTES))
+ .addValidator(StandardValidators.NON_NEGATIVE_INTEGER_VALIDATOR)
Review Comment:
`StandardValidators.createDataSizeBoundsValidator(0, Integer.MAX_VALUE)`.
Kafka stores this setting as an `int`, so plain `DATA_SIZE_VALIDATOR` would
accept a value like `4 GB` that can't be passed to Kafka.
Once the name no longer matches the Kafka key,
`StandardKafkaPropertyProvider` stops forwarding it. `getConsumerProperties`
will need to read it with `asDataSize(DataUnit.B).intValue()` and put it under
`ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG`. `PublishKafka` handles Max
Request Size the same way.
##########
nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/main/java/org/apache/nifi/kafka/service/Kafka3ConnectionService.java:
##########
@@ -128,6 +128,36 @@ public class Kafka3ConnectionService extends
AbstractControllerService implement
.addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
.build();
+ public static final PropertyDescriptor MAX_PARTITION_FETCH_BYTES = new
PropertyDescriptor.Builder()
+ .name("max.partition.fetch.bytes")
+ .displayName("Max Partition Fetch Bytes")
+ .description("""
+ Maximum amount of data in bytes 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(String.valueOf(ConsumerConfig.DEFAULT_MAX_PARTITION_FETCH_BYTES))
+ .addValidator(StandardValidators.NON_NEGATIVE_INTEGER_VALIDATOR)
+ .expressionLanguageSupported(ExpressionLanguageScope.ENVIRONMENT)
+ .build();
+
+ public static final PropertyDescriptor FETCH_MAX_BYTES = new
PropertyDescriptor.Builder()
+ .name("fetch.max.bytes")
+ .displayName("Fetch Max Bytes")
+ .description("""
+ Maximum amount of data in bytes 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(String.valueOf(ConsumerConfig.DEFAULT_FETCH_MAX_BYTES))
+ .addValidator(StandardValidators.NON_NEGATIVE_INTEGER_VALIDATOR)
+ .expressionLanguageSupported(ExpressionLanguageScope.ENVIRONMENT)
Review Comment:
Same comments as above: use only `.name(...)` with a name like `Max Fetch
Size`, a default of `50 MB` (Kafka's default), the bounded data size validator,
and no Expression Language. Also read it explicitly in `getConsumerProperties`
and put it under `ConsumerConfig.FETCH_MAX_BYTES_CONFIG`.
##########
nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/main/java/org/apache/nifi/kafka/service/Kafka3ConnectionService.java:
##########
@@ -176,6 +206,8 @@ public class Kafka3ConnectionService extends
AbstractControllerService implement
SSL_CONTEXT_SERVICE,
TRANSACTION_ISOLATION_LEVEL,
MAX_POLL_RECORDS,
+ MAX_PARTITION_FETCH_BYTES,
Review Comment:
[Claude Opus 5.5] Existing flows may already set `max.partition.fetch.bytes`
and `fetch.max.bytes` as dynamic properties with plain byte values. After the
rename, those flows need `migrateProperties` to carry their values over.
Without it, the old dynamic property is still forwarded to Kafka, and the new
property's default is then applied on top of it, silently replacing the user's
setting.
For each old name, please rename the property to the new name. If the value
is a plain number, also append ` B`, since NiFi's data size format requires a
unit. A value that uses Expression Language can't be converted. Leaving it
unchanged makes the property invalid, which at least shows the user an error
instead of quietly dropping the setting.
`AmazonMSKConnectionService` already overrides `migrateProperties` without
calling `super`, so it will need to call `super.migrateProperties(config)` too.
##########
nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/resources/docs/org.apache.nifi.kafka.processors.ConsumeKafka/additionalDetails.md:
##########
@@ -175,3 +175,10 @@ 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 group has a
lag, every poll is full.
Review Comment:
[Claude Opus 5.5] Could this say "when the consumer falls behind" instead of
"when the consumer group has a lag"? It's a bit easier to read.
##########
nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/test/java/org/apache/nifi/kafka/service/Kafka3ConnectionServiceTest.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.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 org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertSame;
+
+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_BYTES = "262144";
+
+ private static final String FETCH_MAX_BYTES = "10000000";
+
+ 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 testFetchSizeDefaultsInConsumerProperties() {
+ runner.assertValid(service);
+
+ final Properties consumerProperties = getConsumerProperties();
+
+
assertEquals(String.valueOf(ConsumerConfig.DEFAULT_MAX_PARTITION_FETCH_BYTES),
consumerProperties.get(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG));
+ assertEquals(String.valueOf(ConsumerConfig.DEFAULT_FETCH_MAX_BYTES),
consumerProperties.get(ConsumerConfig.FETCH_MAX_BYTES_CONFIG));
+ }
+
+ @Test
+ void testFetchSizeConfiguredInConsumerProperties() {
+ runner.setProperty(service,
Kafka3ConnectionService.MAX_PARTITION_FETCH_BYTES, MAX_PARTITION_FETCH_BYTES);
+ runner.setProperty(service, Kafka3ConnectionService.FETCH_MAX_BYTES,
FETCH_MAX_BYTES);
+ runner.assertValid(service);
+
+ final Properties consumerProperties = getConsumerProperties();
+
+ assertEquals(MAX_PARTITION_FETCH_BYTES,
consumerProperties.get(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG));
+ assertEquals(FETCH_MAX_BYTES,
consumerProperties.get(ConsumerConfig.FETCH_MAX_BYTES_CONFIG));
+ }
+
+ @Test
+ void testFetchSizePropertyNamesResolveToSupportedDescriptors() {
+ runner.setProperty(service,
ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, MAX_PARTITION_FETCH_BYTES);
+ runner.setProperty(service, ConsumerConfig.FETCH_MAX_BYTES_CONFIG,
FETCH_MAX_BYTES);
+ runner.assertValid(service);
+
+ final PropertyDescriptor maxPartitionFetchBytes =
getConfiguredDescriptor(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG);
+ assertSame(Kafka3ConnectionService.MAX_PARTITION_FETCH_BYTES,
maxPartitionFetchBytes);
+ assertFalse(maxPartitionFetchBytes.isDynamic());
+
+ final PropertyDescriptor fetchMaxBytes =
getConfiguredDescriptor(ConsumerConfig.FETCH_MAX_BYTES_CONFIG);
+ assertSame(Kafka3ConnectionService.FETCH_MAX_BYTES, fetchMaxBytes);
+ assertFalse(fetchMaxBytes.isDynamic());
+
+ final Properties consumerProperties = getConsumerProperties();
+
+ assertEquals(MAX_PARTITION_FETCH_BYTES,
consumerProperties.get(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG));
+ assertEquals(FETCH_MAX_BYTES,
consumerProperties.get(ConsumerConfig.FETCH_MAX_BYTES_CONFIG));
+ }
+
+ @Test
+ void testFetchSizeNotInProducerProperties() {
+ runner.setProperty(service,
Kafka3ConnectionService.MAX_PARTITION_FETCH_BYTES, MAX_PARTITION_FETCH_BYTES);
+ runner.setProperty(service, Kafka3ConnectionService.FETCH_MAX_BYTES,
FETCH_MAX_BYTES);
+ runner.assertValid(service);
+
+ final MockConfigurationContext configurationContext =
getConfigurationContext();
+ final Properties clientProperties =
service.getClientProperties(configurationContext);
+ final Properties producerProperties =
service.getProducerProperties(configurationContext, clientProperties);
+
+
assertFalse(producerProperties.containsKey(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG));
+
assertFalse(producerProperties.containsKey(ConsumerConfig.FETCH_MAX_BYTES_CONFIG));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"-1", "1 MB", "unlimited"})
+ void testMaxPartitionFetchBytesNotValid(final String value) {
Review Comment:
[Claude Opus 5.5] This test and the next one are the same test run against
two properties, and they mainly check `StandardValidators`. Once these become
data sizes, `1 MB` will be a valid value, so they won't hold up anyway. Please
remove both.
##########
nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/test/java/org/apache/nifi/kafka/service/Kafka3ConnectionServiceTest.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.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 org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertSame;
+
+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_BYTES = "262144";
+
+ private static final String FETCH_MAX_BYTES = "10000000";
+
+ 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 testFetchSizeDefaultsInConsumerProperties() {
+ runner.assertValid(service);
+
+ final Properties consumerProperties = getConsumerProperties();
+
+
assertEquals(String.valueOf(ConsumerConfig.DEFAULT_MAX_PARTITION_FETCH_BYTES),
consumerProperties.get(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG));
+ assertEquals(String.valueOf(ConsumerConfig.DEFAULT_FETCH_MAX_BYTES),
consumerProperties.get(ConsumerConfig.FETCH_MAX_BYTES_CONFIG));
+ }
+
+ @Test
+ void testFetchSizeConfiguredInConsumerProperties() {
+ runner.setProperty(service,
Kafka3ConnectionService.MAX_PARTITION_FETCH_BYTES, MAX_PARTITION_FETCH_BYTES);
+ runner.setProperty(service, Kafka3ConnectionService.FETCH_MAX_BYTES,
FETCH_MAX_BYTES);
+ runner.assertValid(service);
+
+ final Properties consumerProperties = getConsumerProperties();
+
+ assertEquals(MAX_PARTITION_FETCH_BYTES,
consumerProperties.get(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG));
+ assertEquals(FETCH_MAX_BYTES,
consumerProperties.get(ConsumerConfig.FETCH_MAX_BYTES_CONFIG));
+ }
+
+ @Test
+ void testFetchSizePropertyNamesResolveToSupportedDescriptors() {
+ runner.setProperty(service,
ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, MAX_PARTITION_FETCH_BYTES);
+ runner.setProperty(service, ConsumerConfig.FETCH_MAX_BYTES_CONFIG,
FETCH_MAX_BYTES);
+ runner.assertValid(service);
+
+ final PropertyDescriptor maxPartitionFetchBytes =
getConfiguredDescriptor(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG);
+ assertSame(Kafka3ConnectionService.MAX_PARTITION_FETCH_BYTES,
maxPartitionFetchBytes);
+ assertFalse(maxPartitionFetchBytes.isDynamic());
+
+ final PropertyDescriptor fetchMaxBytes =
getConfiguredDescriptor(ConsumerConfig.FETCH_MAX_BYTES_CONFIG);
+ assertSame(Kafka3ConnectionService.FETCH_MAX_BYTES, fetchMaxBytes);
+ assertFalse(fetchMaxBytes.isDynamic());
+
+ final Properties consumerProperties = getConsumerProperties();
+
+ assertEquals(MAX_PARTITION_FETCH_BYTES,
consumerProperties.get(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG));
+ assertEquals(FETCH_MAX_BYTES,
consumerProperties.get(ConsumerConfig.FETCH_MAX_BYTES_CONFIG));
+ }
+
+ @Test
+ void testFetchSizeNotInProducerProperties() {
Review Comment:
[Claude Opus 5.5] This tests the existing filtering in
`StandardKafkaPropertyProvider`, not anything this PR adds. Once
`getConsumerProperties` puts these values in explicitly, there's nothing left
for it to check. Please remove it.
##########
nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-aws/src/test/java/org/apache/nifi/kafka/service/aws/AmazonMSKConnectionServiceTest.java:
##########
@@ -117,6 +125,34 @@ void testWebIdentityWithTokenProviderAddsProperty() throws
InitializationExcepti
assertInstanceOf(AwsCredentialsProvider.class, provider);
}
+ @Test
+ void testFetchSizePropertiesSupported() {
Review Comment:
[Claude Opus 5.5] Both new tests here repeat the tests in
`Kafka3ConnectionServiceTest`, because the MSK service copies the parent's
property list and calls `super.getConsumerProperties`. This one also checks the
contents of the property list, which is an implementation detail. Please remove
both, along with the new constants and imports.
##########
nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/test/java/org/apache/nifi/kafka/service/Kafka3ConnectionServiceTest.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.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 org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertSame;
+
+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_BYTES = "262144";
+
+ private static final String FETCH_MAX_BYTES = "10000000";
+
+ 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 testFetchSizeDefaultsInConsumerProperties() {
+ runner.assertValid(service);
+
+ final Properties consumerProperties = getConsumerProperties();
+
+
assertEquals(String.valueOf(ConsumerConfig.DEFAULT_MAX_PARTITION_FETCH_BYTES),
consumerProperties.get(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG));
+ assertEquals(String.valueOf(ConsumerConfig.DEFAULT_FETCH_MAX_BYTES),
consumerProperties.get(ConsumerConfig.FETCH_MAX_BYTES_CONFIG));
+ }
+
+ @Test
+ void testFetchSizeConfiguredInConsumerProperties() {
+ runner.setProperty(service,
Kafka3ConnectionService.MAX_PARTITION_FETCH_BYTES, MAX_PARTITION_FETCH_BYTES);
+ runner.setProperty(service, Kafka3ConnectionService.FETCH_MAX_BYTES,
FETCH_MAX_BYTES);
+ runner.assertValid(service);
+
+ final Properties consumerProperties = getConsumerProperties();
+
+ assertEquals(MAX_PARTITION_FETCH_BYTES,
consumerProperties.get(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG));
+ assertEquals(FETCH_MAX_BYTES,
consumerProperties.get(ConsumerConfig.FETCH_MAX_BYTES_CONFIG));
+ }
+
+ @Test
+ void testFetchSizePropertyNamesResolveToSupportedDescriptors() {
+ runner.setProperty(service,
ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, MAX_PARTITION_FETCH_BYTES);
+ runner.setProperty(service, ConsumerConfig.FETCH_MAX_BYTES_CONFIG,
FETCH_MAX_BYTES);
+ runner.assertValid(service);
+
+ final PropertyDescriptor maxPartitionFetchBytes =
getConfiguredDescriptor(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG);
+ assertSame(Kafka3ConnectionService.MAX_PARTITION_FETCH_BYTES,
maxPartitionFetchBytes);
+ assertFalse(maxPartitionFetchBytes.isDynamic());
+
+ final PropertyDescriptor fetchMaxBytes =
getConfiguredDescriptor(ConsumerConfig.FETCH_MAX_BYTES_CONFIG);
+ assertSame(Kafka3ConnectionService.FETCH_MAX_BYTES, fetchMaxBytes);
+ assertFalse(fetchMaxBytes.isDynamic());
+
+ final Properties consumerProperties = getConsumerProperties();
+
+ assertEquals(MAX_PARTITION_FETCH_BYTES,
consumerProperties.get(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG));
+ assertEquals(FETCH_MAX_BYTES,
consumerProperties.get(ConsumerConfig.FETCH_MAX_BYTES_CONFIG));
+ }
+
+ @Test
+ void testFetchSizeNotInProducerProperties() {
+ runner.setProperty(service,
Kafka3ConnectionService.MAX_PARTITION_FETCH_BYTES, MAX_PARTITION_FETCH_BYTES);
+ runner.setProperty(service, Kafka3ConnectionService.FETCH_MAX_BYTES,
FETCH_MAX_BYTES);
+ runner.assertValid(service);
+
+ final MockConfigurationContext configurationContext =
getConfigurationContext();
+ final Properties clientProperties =
service.getClientProperties(configurationContext);
+ final Properties producerProperties =
service.getProducerProperties(configurationContext, clientProperties);
+
+
assertFalse(producerProperties.containsKey(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG));
+
assertFalse(producerProperties.containsKey(ConsumerConfig.FETCH_MAX_BYTES_CONFIG));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"-1", "1 MB", "unlimited"})
+ void testMaxPartitionFetchBytesNotValid(final String value) {
+ runner.setProperty(service,
Kafka3ConnectionService.MAX_PARTITION_FETCH_BYTES, value);
+
+ runner.assertNotValid(service);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"-1", "1 MB", "unlimited"})
+ void testFetchMaxBytesNotValid(final String value) {
+ runner.setProperty(service, Kafka3ConnectionService.FETCH_MAX_BYTES,
value);
+
+ runner.assertNotValid(service);
+ }
+
+ private PropertyDescriptor getConfiguredDescriptor(final String
propertyName) {
+ final MockProcessContext processContext = (MockProcessContext)
runner.getProcessContext();
+ return
processContext.getConfiguration(SERVICE_ID).getProperties().keySet().stream()
+ .filter(descriptor ->
descriptor.getName().equals(propertyName))
+ .findFirst()
+ .orElseThrow();
+ }
+
+ private Properties getConsumerProperties() {
+ final MockConfigurationContext configurationContext =
getConfigurationContext();
+ final Properties clientProperties =
service.getClientProperties(configurationContext);
+ return service.getConsumerProperties(configurationContext,
clientProperties);
+ }
+
+ private MockConfigurationContext getConfigurationContext() {
+ final MockProcessContext processContext = (MockProcessContext)
runner.getProcessContext();
+
+ // The framework provides default values for all supported properties,
while the runner provides only configured values
+ final Map<PropertyDescriptor, String> properties = new
LinkedHashMap<>();
Review Comment:
[Claude Opus 5.5] A `HashMap` is enough here, since nothing depends on the
order.
##########
nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/test/java/org/apache/nifi/kafka/service/Kafka3ConnectionServiceTest.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.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 org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertSame;
+
+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_BYTES = "262144";
+
+ private static final String FETCH_MAX_BYTES = "10000000";
+
+ 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 testFetchSizeDefaultsInConsumerProperties() {
+ runner.assertValid(service);
+
+ final Properties consumerProperties = getConsumerProperties();
+
+
assertEquals(String.valueOf(ConsumerConfig.DEFAULT_MAX_PARTITION_FETCH_BYTES),
consumerProperties.get(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG));
+ assertEquals(String.valueOf(ConsumerConfig.DEFAULT_FETCH_MAX_BYTES),
consumerProperties.get(ConsumerConfig.FETCH_MAX_BYTES_CONFIG));
+ }
+
+ @Test
+ void testFetchSizeConfiguredInConsumerProperties() {
+ runner.setProperty(service,
Kafka3ConnectionService.MAX_PARTITION_FETCH_BYTES, MAX_PARTITION_FETCH_BYTES);
+ runner.setProperty(service, Kafka3ConnectionService.FETCH_MAX_BYTES,
FETCH_MAX_BYTES);
+ runner.assertValid(service);
+
+ final Properties consumerProperties = getConsumerProperties();
+
+ assertEquals(MAX_PARTITION_FETCH_BYTES,
consumerProperties.get(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG));
+ assertEquals(FETCH_MAX_BYTES,
consumerProperties.get(ConsumerConfig.FETCH_MAX_BYTES_CONFIG));
+ }
+
+ @Test
+ void testFetchSizePropertyNamesResolveToSupportedDescriptors() {
Review Comment:
[Claude Opus 5.5] This mostly checks the framework's lookup of a property by
name, and its last assertions repeat
`testFetchSizeConfiguredInConsumerProperties`. After the rename, it's the right
place for a migration test instead.
Build a `MockPropertyConfiguration` with the old dynamic names and plain
byte values, then call `service.migrateProperties(...)`. Check that the new
names hold the values with ` B` appended. Please also include one Expression
Language value to show it's left unchanged. `TestRunner.migrateProperties()`
only works for processors, so the service has to be called directly.
##########
nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/main/java/org/apache/nifi/kafka/service/Kafka3ConnectionService.java:
##########
@@ -128,6 +128,36 @@ public class Kafka3ConnectionService extends
AbstractControllerService implement
.addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
.build();
+ public static final PropertyDescriptor MAX_PARTITION_FETCH_BYTES = new
PropertyDescriptor.Builder()
+ .name("max.partition.fetch.bytes")
+ .displayName("Max Partition Fetch Bytes")
+ .description("""
+ Maximum amount of data in bytes 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(String.valueOf(ConsumerConfig.DEFAULT_MAX_PARTITION_FETCH_BYTES))
Review Comment:
`1 MB`. That's exactly Kafka's default, because NiFi counts 1 KB as 1024
bytes, so existing flows behave the same.
##########
nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/test/java/org/apache/nifi/kafka/service/Kafka3ConnectionServiceTest.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.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 org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertSame;
+
+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_BYTES = "262144";
+
+ private static final String FETCH_MAX_BYTES = "10000000";
+
+ 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 testFetchSizeDefaultsInConsumerProperties() {
Review Comment:
[Claude Opus 5.5] This and the next test can be combined: check the
defaults, then set values on the same runner and check again.
##########
nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/resources/docs/org.apache.nifi.kafka.processors.ConsumeKafka/additionalDetails.md:
##########
@@ -175,3 +175,10 @@ 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 group has a
lag, every poll is full.
+
+Memory usage can be limited by lowering the 'Max Partition Fetch Bytes'
property of the Kafka Connection Service.
Review Comment:
[Claude Opus 5.5] Please update this to the new property name. It's also
worth mentioning Max Poll Records, which caps how many records a single poll
returns and so limits the same memory. And it should note the trade-off:
smaller fetches mean more requests to the broker and lower throughput.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
