You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@nifi.apache.org by "ing-mattioni (via GitHub)" <gi...@apache.org> on 2026/05/13 10:22:24 UTC
[PR] NIFI-15934 Handle oversized FlowFiles and empty header keys in PublishAMQP [nifi]
ing-mattioni opened a new pull request, #11242:
URL: https://github.com/apache/nifi/pull/11242
…hAMQP
<!-- 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. -->
# Summary
[NIFI-15934](https://issues.apache.org/jira/browse/NIFI-15934)
This pull request updates `PublishAMQP` to improve handling of oversized FlowFiles and AMQP header attributes.
`PublishAMQP` reads FlowFile content into a byte array before publishing an AMQP message. Since Java byte arrays require an `int` length, FlowFiles larger than `Integer.MAX_VALUE` cannot be published using this path. This change validates the FlowFile size before attempting byte-array allocation and routes oversized FlowFiles to `failure` with penalization.
This pull request also updates AMQP header parsing to ignore empty header keys instead of adding them to the AMQP headers map.
P.S. The existing unit test for AMQP header parsing was extended to cover empty header keys. The oversized FlowFile guard is a defensive validation before byte-array allocation; additional targeted coverage may require avoiding real multi-GB content allocation.
## Changes
- Validates FlowFile size before extracting message content into a byte array
- Routes FlowFiles larger than `Integer.MAX_VALUE` to `failure` with penalization
- Uses `Math.toIntExact()` when converting FlowFile size to byte array length
- Ignores empty AMQP header keys when parsing AMQP header attributes
- Extends AMQP header parsing test coverage for empty header keys
# Tracking
Please complete the following tracking steps prior to pull request creation.
### Issue Tracking
- [x] [Apache NiFi Jira](https://issues.apache.org/jira/browse/NIFI-15934) issue created
### Pull Request Tracking
- [x] Pull Request title starts with Apache NiFi Jira issue number, such as `NIFI-00000`
- [x] Pull Request commit message starts with Apache NiFi Jira issue number, as such `NIFI-00000`
- [x] Pull request contains commits signed with a registered key indicating `Verified` status
### Pull Request Formatting
- [x] Pull Request based on current revision of the `main` branch
- [x] Pull Request refers to a feature branch with one commit containing changes
# Verification
Please indicate the verification steps performed prior to pull request creation.
### Build
- [ ] Build completed using `./mvnw clean install -P contrib-check`
- [ ] JDK 21
- [ ] JDK 25
Module-level verification performed:
`.\mvnw.cmd -pl :nifi-amqp-processors -am test`
Additional verification performed:
`.\mvnw.cmd -pl :nifi-amqp-processors -am -P contrib-check install -DskipTests`
`.\mvnw.cmd -pl :nifi-amqp-nar -am install -DskipTests`
### Licensing
- [x] New dependencies are compatible with the Apache License 2.0
- [x] New dependencies are documented in applicable `LICENSE` and `NOTICE` files
No new dependencies were added.
### Documentation
- [x] Documentation formatting appears as expected in rendered files
No user-facing documentation files were changed.
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15934 Handle oversized FlowFiles and empty header keys in PublishAMQP [nifi]
Posted by "ing-mattioni (via GitHub)" <gi...@apache.org>.
ing-mattioni commented on code in PR #11242:
URL: https://github.com/apache/nifi/pull/11242#discussion_r3257465451
##########
nifi-extension-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/main/java/org/apache/nifi/amqp/processors/PublishAMQP.java:
##########
@@ -317,9 +323,18 @@ private Map<String, Object> validateAMQPHeaderProperty(final String amqpPropValu
for (String strEntry : strEntries) {
final String[] kv = strEntry.split("=", -1); // without using limit, trailing delimiter would be ignored
if (kv.length == 2) {
- headers.put(kv[0].trim(), kv[1].trim());
+ final String key = kv[0].trim();
+ if (key.isEmpty()) {
+ getLogger().warn("Ignoring AMQP header property with empty key: {}", strEntry);
+ continue;
+ }
+ headers.put(key, kv[1].trim());
} else if (kv.length == 1) {
- headers.put(kv[0].trim(), null);
+ final String key = kv[0].trim();
+ if (key.isEmpty()) {
+ continue;
+ }
+ headers.put(key, null);
Review Comment:
Extracted header insertion into a helper so empty keys are handled consistently for both key=value and bare-key entries.
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15934 Handle oversized FlowFiles and empty header keys in PublishAMQP [nifi]
Posted by "exceptionfactory (via GitHub)" <gi...@apache.org>.
exceptionfactory commented on code in PR #11242:
URL: https://github.com/apache/nifi/pull/11242#discussion_r3268000454
##########
nifi-extension-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/main/java/org/apache/nifi/amqp/processors/PublishAMQP.java:
##########
@@ -95,6 +97,14 @@ public class PublishAMQP extends AbstractAMQPProcessor<AMQPPublisher> {
.expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
.addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
.build();
+ public static final PropertyDescriptor MAX_OUTBOUND_MESSAGE_BODY_SIZE = new PropertyDescriptor.Builder()
+ .name("Max Outbound Message Body Size")
+ .description("Maximum body size of outbound messages. FlowFiles larger than this value are routed to failure without loading content into memory.")
+ .required(true)
+ .defaultValue("64 MB")
Review Comment:
Is there a general published maximum size of AMQP messages, or some reference for using 64 MB is the default value?
##########
nifi-extension-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/main/java/org/apache/nifi/amqp/processors/PublishAMQP.java:
##########
@@ -95,6 +97,14 @@ public class PublishAMQP extends AbstractAMQPProcessor<AMQPPublisher> {
.expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
.addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
.build();
+ public static final PropertyDescriptor MAX_OUTBOUND_MESSAGE_BODY_SIZE = new PropertyDescriptor.Builder()
+ .name("Max Outbound Message Body Size")
Review Comment:
I recommend naming this `Maximum Input FlowFile Size` since that it what is controls.
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15934 Handle oversized FlowFiles and empty header keys in PublishAMQP [nifi]
Posted by "ing-mattioni (via GitHub)" <gi...@apache.org>.
ing-mattioni commented on PR #11242:
URL: https://github.com/apache/nifi/pull/11242#issuecomment-4542005384
Conflicts in PublishAMQP class resolved
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15934 Handle oversized FlowFiles and empty header keys in PublishAMQP [nifi]
Posted by "exceptionfactory (via GitHub)" <gi...@apache.org>.
exceptionfactory merged PR #11242:
URL: https://github.com/apache/nifi/pull/11242
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15934 Handle oversized FlowFiles and empty header keys in PublishAMQP [nifi]
Posted by "ing-mattioni (via GitHub)" <gi...@apache.org>.
ing-mattioni commented on code in PR #11242:
URL: https://github.com/apache/nifi/pull/11242#discussion_r3257464096
##########
nifi-extension-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/main/java/org/apache/nifi/amqp/processors/PublishAMQP.java:
##########
@@ -166,6 +166,12 @@ protected void processResource(final Connection connection, final AMQPPublisher
return;
}
+ if (flowFile.getSize() > Integer.MAX_VALUE) {
+ getLogger().error("FlowFile {} with size {} bytes is too large to publish as an AMQP message", flowFile, flowFile.getSize());
+ session.transfer(session.penalize(flowFile), REL_FAILURE);
Review Comment:
Removed penalization when routing to failure and added a regression assertion.
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15934 Handle oversized FlowFiles and empty header keys in PublishAMQP [nifi]
Posted by "pvillard31 (via GitHub)" <gi...@apache.org>.
pvillard31 commented on code in PR #11242:
URL: https://github.com/apache/nifi/pull/11242#discussion_r3252716021
##########
nifi-extension-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/main/java/org/apache/nifi/amqp/processors/PublishAMQP.java:
##########
@@ -220,7 +226,7 @@ public void migrateProperties(final PropertyConfiguration config) {
* Extracts contents of the {@link FlowFile} as byte array.
*/
private byte[] extractMessage(final FlowFile flowFile, ProcessSession session) {
- final byte[] messageContent = new byte[(int) flowFile.getSize()];
+ final byte[] messageContent = new byte[Math.toIntExact(flowFile.getSize())];
Review Comment:
is `Math.toIntExact` required since we filter out larger flowfiles before?
##########
nifi-extension-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/main/java/org/apache/nifi/amqp/processors/PublishAMQP.java:
##########
@@ -317,9 +323,18 @@ private Map<String, Object> validateAMQPHeaderProperty(final String amqpPropValu
for (String strEntry : strEntries) {
final String[] kv = strEntry.split("=", -1); // without using limit, trailing delimiter would be ignored
if (kv.length == 2) {
- headers.put(kv[0].trim(), kv[1].trim());
+ final String key = kv[0].trim();
+ if (key.isEmpty()) {
+ getLogger().warn("Ignoring AMQP header property with empty key: {}", strEntry);
+ continue;
+ }
+ headers.put(key, kv[1].trim());
} else if (kv.length == 1) {
- headers.put(kv[0].trim(), null);
+ final String key = kv[0].trim();
+ if (key.isEmpty()) {
+ continue;
+ }
+ headers.put(key, null);
Review Comment:
Should the two empty-key branches log at the same level (both warn, or both silent), and could the trim plus empty check be extracted into a small helper so the behavior is identical for key=value and bare-key entries?
##########
nifi-extension-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/main/java/org/apache/nifi/amqp/processors/PublishAMQP.java:
##########
@@ -166,6 +166,12 @@ protected void processResource(final Connection connection, final AMQPPublisher
return;
}
+ if (flowFile.getSize() > Integer.MAX_VALUE) {
+ getLogger().error("FlowFile {} with size {} bytes is too large to publish as an AMQP message", flowFile, flowFile.getSize());
+ session.transfer(session.penalize(flowFile), REL_FAILURE);
Review Comment:
we usually don't penalize when sending into the failure relationship unless we believe that the failure relationship is used as a self-loop on the processor (but this is not a good practice in terms of flow design given that we have the retry at framework level)
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15934 Handle oversized FlowFiles and empty header keys in PublishAMQP [nifi]
Posted by "ing-mattioni (via GitHub)" <gi...@apache.org>.
ing-mattioni commented on code in PR #11242:
URL: https://github.com/apache/nifi/pull/11242#discussion_r3814926700
##########
nifi-extension-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/main/java/org/apache/nifi/amqp/processors/PublishAMQP.java:
##########
@@ -75,6 +76,7 @@
@ReadsAttribute(attribute = AbstractAMQPProcessor.AMQP_CLUSTER_ID_ATTRIBUTE, description = "The ID of the AMQP Cluster"),
})
public class PublishAMQP extends AbstractAMQPProcessor<AMQPPublisher> {
+ private static final long MAXIMUM_INPUT_FLOWFILE_SIZE_LIMIT = 128 * 1024 * 1024L;
Review Comment:
Thanks for pointing this out. I updated the maximum allowed value from 128 MB to 512 MB.
##########
nifi-extension-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/main/java/org/apache/nifi/amqp/processors/PublishAMQP.java:
##########
@@ -108,6 +110,16 @@ public class PublishAMQP extends AbstractAMQPProcessor<AMQPPublisher> {
.allowableValues(DeliveryGuarantee.class)
.defaultValue(DeliveryGuarantee.AT_MOST_ONCE)
.build();
+ public static final PropertyDescriptor MAXIMUM_INPUT_FLOWFILE_SIZE = new PropertyDescriptor.Builder()
+ .name("Maximum Input FlowFile Size")
+ .description("Maximum size of an input FlowFile that will be read into memory before publishing. PublishAMQP reads FlowFile content into a byte array "
+ + "before publishing, so FlowFiles larger than this value are routed to failure before content is read. Configure this value according to "
+ + "broker limits and available JVM memory.")
Review Comment:
Thanks for the suggestion. I updated the description to use a multiline text block.
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15934 Handle oversized FlowFiles and empty header keys in PublishAMQP [nifi]
Posted by "ing-mattioni (via GitHub)" <gi...@apache.org>.
ing-mattioni commented on code in PR #11242:
URL: https://github.com/apache/nifi/pull/11242#discussion_r3257462840
##########
nifi-extension-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/main/java/org/apache/nifi/amqp/processors/PublishAMQP.java:
##########
@@ -220,7 +226,7 @@ public void migrateProperties(final PropertyConfiguration config) {
* Extracts contents of the {@link FlowFile} as byte array.
*/
private byte[] extractMessage(final FlowFile flowFile, ProcessSession session) {
- final byte[] messageContent = new byte[(int) flowFile.getSize()];
+ final byte[] messageContent = new byte[Math.toIntExact(flowFile.getSize())];
Review Comment:
Resolved by checking against the configured size limit before allocation, so the remaining cast is bounded.
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15934 Handle oversized FlowFiles and empty header keys in PublishAMQP [nifi]
Posted by "ing-mattioni (via GitHub)" <gi...@apache.org>.
ing-mattioni commented on PR #11242:
URL: https://github.com/apache/nifi/pull/11242#issuecomment-4621084334
Rebased now with the latest main version after validation and build failed
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15934 Handle oversized FlowFiles and empty header keys in PublishAMQP [nifi]
Posted by "ing-mattioni (via GitHub)" <gi...@apache.org>.
ing-mattioni commented on PR #11242:
URL: https://github.com/apache/nifi/pull/11242#issuecomment-5352982786
@exceptionfactory Thanks for pointing this out. I rebased the branch onto the latest main, signed all commits, and force-pushed the updated branch. GitHub now reports both commits as verified.
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15934 Handle oversized FlowFiles and empty header keys in PublishAMQP [nifi]
Posted by "ing-mattioni (via GitHub)" <gi...@apache.org>.
ing-mattioni commented on PR #11242:
URL: https://github.com/apache/nifi/pull/11242#issuecomment-4516421374
Rebased now with the latest main version after validation and build failed
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15934 Handle oversized FlowFiles and empty header keys in PublishAMQP [nifi]
Posted by "ing-mattioni (via GitHub)" <gi...@apache.org>.
ing-mattioni commented on PR #11242:
URL: https://github.com/apache/nifi/pull/11242#issuecomment-4506871813
Addressed in the latest update:
- I renamed the property to `Maximum Input FlowFile Size` to clarify that it controls the maximum input FlowFile size before `PublishAMQP` reads the content into memory.
- I also updated the default value and description so the property is described as a NiFi-side defensive memory guard rather than a general AMQP protocol limit. The value should be configured according to broker limits and available JVM memory.
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15934 Handle oversized FlowFiles and empty header keys in PublishAMQP [nifi]
Posted by "exceptionfactory (via GitHub)" <gi...@apache.org>.
exceptionfactory commented on code in PR #11242:
URL: https://github.com/apache/nifi/pull/11242#discussion_r3706810183
##########
nifi-extension-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/main/java/org/apache/nifi/amqp/processors/PublishAMQP.java:
##########
@@ -75,6 +76,7 @@
@ReadsAttribute(attribute = AbstractAMQPProcessor.AMQP_CLUSTER_ID_ATTRIBUTE, description = "The ID of the AMQP Cluster"),
})
public class PublishAMQP extends AbstractAMQPProcessor<AMQPPublisher> {
+ private static final long MAXIMUM_INPUT_FLOWFILE_SIZE_LIMIT = 128 * 1024 * 1024L;
Review Comment:
Based on the earlier conversation, it sounds like the maximum value should be 512 MB, not 128 MB.
##########
nifi-extension-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/main/java/org/apache/nifi/amqp/processors/PublishAMQP.java:
##########
@@ -108,6 +110,16 @@ public class PublishAMQP extends AbstractAMQPProcessor<AMQPPublisher> {
.allowableValues(DeliveryGuarantee.class)
.defaultValue(DeliveryGuarantee.AT_MOST_ONCE)
.build();
+ public static final PropertyDescriptor MAXIMUM_INPUT_FLOWFILE_SIZE = new PropertyDescriptor.Builder()
+ .name("Maximum Input FlowFile Size")
+ .description("Maximum size of an input FlowFile that will be read into memory before publishing. PublishAMQP reads FlowFile content into a byte array "
+ + "before publishing, so FlowFiles larger than this value are routed to failure before content is read. Configure this value according to "
+ + "broker limits and available JVM memory.")
Review Comment:
A multline string can be used
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15934 Handle oversized FlowFiles and empty header keys in PublishAMQP [nifi]
Posted by "ing-mattioni (via GitHub)" <gi...@apache.org>.
ing-mattioni commented on code in PR #11242:
URL: https://github.com/apache/nifi/pull/11242#discussion_r3823520956
##########
nifi-extension-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/test/java/org/apache/nifi/amqp/processors/PublishAMQPTest.java:
##########
@@ -181,7 +210,42 @@ public void validateFailedPublishAndTransferToFailure() {
runner.run();
assertTrue(runner.getFlowFilesForRelationship(PublishAMQP.REL_SUCCESS).isEmpty());
- assertNotNull(runner.getFlowFilesForRelationship(PublishAMQP.REL_FAILURE).getFirst());
+ final MockFlowFile failureFlowFile = runner.getFlowFilesForRelationship(PublishAMQP.REL_FAILURE).getFirst();
+ assertNotNull(failureFlowFile);
+ assertFalse(failureFlowFile.isPenalized());
+ runner.assertPenalizeCount(0);
+ }
+
+ @Test
+ public void validateOversizedFlowFileTransferredToFailureWithoutPublishing() throws Exception {
+ setConnectionProperties(runner);
+ runner.setProperty(PublishAMQP.MAXIMUM_INPUT_FLOWFILE_SIZE, "4 B");
+
+ runner.enqueue("Hello".getBytes());
+
+ runner.run();
+
+ assertTrue(runner.getFlowFilesForRelationship(PublishAMQP.REL_SUCCESS).isEmpty());
+ final MockFlowFile failureFlowFile = runner.getFlowFilesForRelationship(PublishAMQP.REL_FAILURE).getFirst();
+ assertNotNull(failureFlowFile);
+ assertFalse(failureFlowFile.isPenalized());
+ runner.assertPenalizeCount(0);
+
+ final Channel channel = pubProc.getConnection().createChannel();
+ assertNull(channel.basicGet("queue1", true));
+ }
+
+ @Test
+ public void validateMaximumInputFlowFileSizeProperty() {
+ assertEquals("Maximum Input FlowFile Size", PublishAMQP.MAXIMUM_INPUT_FLOWFILE_SIZE.getName());
+ assertEquals("128 MB", PublishAMQP.MAXIMUM_INPUT_FLOWFILE_SIZE.getDefaultValue());
+
+ setConnectionProperties(runner);
+ runner.setProperty(PublishAMQP.MAXIMUM_INPUT_FLOWFILE_SIZE, "128 MB");
+ runner.assertValid();
+
+ runner.setProperty(PublishAMQP.MAXIMUM_INPUT_FLOWFILE_SIZE, "129 MB");
+ runner.assertNotValid();
}
Review Comment:
Removed the property descriptor test method as suggested and confirmed that all remaining PublishAMQPTest tests pass.
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15934 Handle oversized FlowFiles and empty header keys in PublishAMQP [nifi]
Posted by "ing-mattioni (via GitHub)" <gi...@apache.org>.
ing-mattioni commented on code in PR #11242:
URL: https://github.com/apache/nifi/pull/11242#discussion_r3280185023
##########
nifi-extension-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/main/java/org/apache/nifi/amqp/processors/PublishAMQP.java:
##########
@@ -95,6 +97,14 @@ public class PublishAMQP extends AbstractAMQPProcessor<AMQPPublisher> {
.expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
.addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
.build();
+ public static final PropertyDescriptor MAX_OUTBOUND_MESSAGE_BODY_SIZE = new PropertyDescriptor.Builder()
+ .name("Max Outbound Message Body Size")
+ .description("Maximum body size of outbound messages. FlowFiles larger than this value are routed to failure without loading content into memory.")
+ .required(true)
+ .defaultValue("64 MB")
Review Comment:
Good point. I updated the change to avoid treating 64 MB as a general AMQP maximum.
RabbitMQ documents `max_message_size` as configurable, with a current documented default of 128 MiB and a maximum allowed value of 512 MiB. I changed the property to use a 128 MB default and described it as a NiFi-side defensive memory guard rather than a protocol-level AMQP limit. The property description now recommends configuring the value according to broker limits and available JVM memory.
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15934 Handle oversized FlowFiles and empty header keys in PublishAMQP [nifi]
Posted by "ing-mattioni (via GitHub)" <gi...@apache.org>.
ing-mattioni commented on PR #11242:
URL: https://github.com/apache/nifi/pull/11242#issuecomment-4475879325
Thank you @pvillard31 and @exceptionfactory , I made the following changes:
- Added configurable Max Outbound Message Body Size in nifi-extension-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/main/java/org/apache/nifi/amqp/processors/PublishAMQP.java:100, default/bounded at 64 MB.
- Oversized FlowFiles now route to failure before content is read into memory.
- Removed penalization when routing publish failures to failure.
- Centralized AMQP header insertion with consistent empty-key skip/warn behavior.
- Added tests for oversized FlowFiles, non-penalized failures, and empty header keys in nifi-extension-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/test/java/org/apache/nifi/amqp/processors/PublishAMQPTest.java:177.
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15934 Handle oversized FlowFiles and empty header keys in PublishAMQP [nifi]
Posted by "ing-mattioni (via GitHub)" <gi...@apache.org>.
ing-mattioni commented on PR #11242:
URL: https://github.com/apache/nifi/pull/11242#issuecomment-4477036258
Addressed the Checkstyle failure in the latest update.
The issue was caused by tab characters in `PublishAMQP.java`, which resulted in `FileTabCharacter` and indentation violations. I replaced the tabs with spaces, reran the local validation, and force-pushed the updated signed commit.
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15934 Handle oversized FlowFiles and empty header keys in PublishAMQP [nifi]
Posted by "exceptionfactory (via GitHub)" <gi...@apache.org>.
exceptionfactory commented on code in PR #11242:
URL: https://github.com/apache/nifi/pull/11242#discussion_r3821990479
##########
nifi-extension-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/test/java/org/apache/nifi/amqp/processors/PublishAMQPTest.java:
##########
@@ -181,7 +210,42 @@ public void validateFailedPublishAndTransferToFailure() {
runner.run();
assertTrue(runner.getFlowFilesForRelationship(PublishAMQP.REL_SUCCESS).isEmpty());
- assertNotNull(runner.getFlowFilesForRelationship(PublishAMQP.REL_FAILURE).getFirst());
+ final MockFlowFile failureFlowFile = runner.getFlowFilesForRelationship(PublishAMQP.REL_FAILURE).getFirst();
+ assertNotNull(failureFlowFile);
+ assertFalse(failureFlowFile.isPenalized());
+ runner.assertPenalizeCount(0);
+ }
+
+ @Test
+ public void validateOversizedFlowFileTransferredToFailureWithoutPublishing() throws Exception {
+ setConnectionProperties(runner);
+ runner.setProperty(PublishAMQP.MAXIMUM_INPUT_FLOWFILE_SIZE, "4 B");
+
+ runner.enqueue("Hello".getBytes());
+
+ runner.run();
+
+ assertTrue(runner.getFlowFilesForRelationship(PublishAMQP.REL_SUCCESS).isEmpty());
+ final MockFlowFile failureFlowFile = runner.getFlowFilesForRelationship(PublishAMQP.REL_FAILURE).getFirst();
+ assertNotNull(failureFlowFile);
+ assertFalse(failureFlowFile.isPenalized());
+ runner.assertPenalizeCount(0);
+
+ final Channel channel = pubProc.getConnection().createChannel();
+ assertNull(channel.basicGet("queue1", true));
+ }
+
+ @Test
+ public void validateMaximumInputFlowFileSizeProperty() {
+ assertEquals("Maximum Input FlowFile Size", PublishAMQP.MAXIMUM_INPUT_FLOWFILE_SIZE.getName());
+ assertEquals("128 MB", PublishAMQP.MAXIMUM_INPUT_FLOWFILE_SIZE.getDefaultValue());
+
+ setConnectionProperties(runner);
+ runner.setProperty(PublishAMQP.MAXIMUM_INPUT_FLOWFILE_SIZE, "128 MB");
+ runner.assertValid();
+
+ runner.setProperty(PublishAMQP.MAXIMUM_INPUT_FLOWFILE_SIZE, "129 MB");
+ runner.assertNotValid();
}
Review Comment:
This test method is failing, but it should be removed since it is simply evaluating property descriptor values
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15934 Handle oversized FlowFiles and empty header keys in PublishAMQP [nifi]
Posted by "ing-mattioni (via GitHub)" <gi...@apache.org>.
ing-mattioni commented on PR #11242:
URL: https://github.com/apache/nifi/pull/11242#issuecomment-4477692267
The remaining test failure appears to be unrelated to this PR. The failed test is in `nifi-aws-processors`:
`org.apache.nifi.processors.aws.kinesis.stream.pause.TestStandardRecordProcessorBlocker.testBlockAndUnblock`
The failure is a timeout while waiting for an asynchronous condition, while this PR only changes `nifi-amqp-processors`.
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15934 Handle oversized FlowFiles and empty header keys in PublishAMQP [nifi]
Posted by "ing-mattioni (via GitHub)" <gi...@apache.org>.
ing-mattioni commented on code in PR #11242:
URL: https://github.com/apache/nifi/pull/11242#discussion_r3257461577
##########
nifi-extension-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/main/java/org/apache/nifi/amqp/processors/PublishAMQP.java:
##########
@@ -166,6 +166,12 @@ protected void processResource(final Connection connection, final AMQPPublisher
return;
}
+ if (flowFile.getSize() > Integer.MAX_VALUE) {
Review Comment:
Addressed by adding a configurable outbound body-size property with a 64 MB default and upper bound, so oversized FlowFiles are routed to failure before reading content into memory.
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15934 Handle oversized FlowFiles and empty header keys in PublishAMQP [nifi]
Posted by "exceptionfactory (via GitHub)" <gi...@apache.org>.
exceptionfactory commented on code in PR #11242:
URL: https://github.com/apache/nifi/pull/11242#discussion_r3253204063
##########
nifi-extension-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/main/java/org/apache/nifi/amqp/processors/PublishAMQP.java:
##########
@@ -166,6 +166,12 @@ protected void processResource(final Connection connection, final AMQPPublisher
return;
}
+ if (flowFile.getSize() > Integer.MAX_VALUE) {
Review Comment:
This does not seem like the right approach, the `Integer.MAX_VALUE` is a very large limit, but there should be other more sensible boundaries, probably lower.
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15934 Handle oversized FlowFiles and empty header keys in PublishAMQP [nifi]
Posted by "ing-mattioni (via GitHub)" <gi...@apache.org>.
ing-mattioni commented on code in PR #11242:
URL: https://github.com/apache/nifi/pull/11242#discussion_r3280179375
##########
nifi-extension-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/main/java/org/apache/nifi/amqp/processors/PublishAMQP.java:
##########
@@ -95,6 +97,14 @@ public class PublishAMQP extends AbstractAMQPProcessor<AMQPPublisher> {
.expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
.addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
.build();
+ public static final PropertyDescriptor MAX_OUTBOUND_MESSAGE_BODY_SIZE = new PropertyDescriptor.Builder()
+ .name("Max Outbound Message Body Size")
Review Comment:
Thanks, that makes sense. I renamed the property to `Maximum Input FlowFile Size` to make it clear that the limit applies to the input FlowFile before `PublishAMQP` reads the content into memory.
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org