wojtre commented on code in PR #11894:
URL: https://github.com/apache/nifi/pull/11894#discussion_r4494041211
##########
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:
Added `migrateProperties` to `Kafka3ConnectionService`. It renames
`max.partition.fetch.bytes` and `fetch.max.bytes` to the new property names and
appends ` B` when the value is a plain number. Any other value, such as
Expression Language or a parameter reference, is moved unchanged, so the
service becomes invalid instead of silently replacing the setting.
`AmazonMSKConnectionService.migrateProperties` now calls
`super.migrateProperties(config)` first.
##########
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:
Changed.
##########
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:
Updated. The section now names `Max Partition Fetch Size` and `Max Poll
Records`, and notes the trade-off: smaller fetches mean more requests to the
broker, and fewer records per poll mean smaller FlowFiles.
##########
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:
Combined into `testFetchSizeInConsumerProperties`. It checks the defaults,
then sets `256 KB` and `10 MB` on the same runner and checks the byte values.
##########
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:
Replaced with `testMigrateProperties`. It calls `migrateProperties` on a
`MockPropertyConfiguration` with the old names and plain byte values, and
checks that the new names hold the values with ` B` appended.
`testMigratePropertiesExpressionLanguageNotConverted` covers an Expression
Language value, which is moved to the new name unchanged.
--
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]
