wojtre commented on code in PR #11894:
URL: https://github.com/apache/nifi/pull/11894#discussion_r4494039057
##########
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:
Done. Both properties now use only `.name(...)`: `Max Partition Fetch Size`
and `Max Fetch Size`. They accept data sizes such as `256 KB` or `1 MB`. The
constants are renamed to `MAX_PARTITION_FETCH_SIZE` and `MAX_FETCH_SIZE` to
match.
##########
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:
Done. The default is now `1 MB`, which equals
`ConsumerConfig.DEFAULT_MAX_PARTITION_FETCH_BYTES`. The test checks that value
in the consumer properties.
##########
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:
Removed from both properties.
##########
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:
Done. Both properties use
`StandardValidators.createDataSizeBoundsValidator(0, Integer.MAX_VALUE)`.
`getConsumerProperties` reads them with `asDataSize(DataUnit.B).intValue()` and
puts them under `ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG` and
`ConsumerConfig.FETCH_MAX_BYTES_CONFIG`, the same way Max Request Size is
handled.
##########
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:
Done: `Max Fetch Size` with a default of `50 MB` (equal to
`ConsumerConfig.DEFAULT_FETCH_MAX_BYTES`), the bounded data size validator, and
no Expression Language. `getConsumerProperties` puts it under
`ConsumerConfig.FETCH_MAX_BYTES_CONFIG`.
--
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]
