This is an automated email from the ASF dual-hosted git repository. joewitt pushed a commit to branch NIFI-16315 in repository https://gitbox.apache.org/repos/asf/nifi.git
commit 7671cc01c5739a56c657d3736f5a0d84c7e0e8fa Author: Joseph Witt <[email protected]> AuthorDate: Mon Sep 7 16:59:42 2026 -0700 NIFI-16315 Reject oversized PutUDP FlowFiles before copying content Co-authored-by: Cursor <[email protected]> --- .../org/apache/nifi/processors/standard/PutUDP.java | 19 ++++++++++++++++--- .../apache/nifi/processors/standard/TestPutUDP.java | 14 +++++++++----- 2 files changed, 25 insertions(+), 8 deletions(-) diff --git a/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutUDP.java b/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutUDP.java index b36129c62ab..9be8c7c83e8 100644 --- a/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutUDP.java +++ b/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutUDP.java @@ -38,15 +38,20 @@ import java.io.IOException; import java.io.InputStream; import java.util.concurrent.TimeUnit; -@CapabilityDescription("The PutUDP processor receives a FlowFile and packages the FlowFile content into a single UDP datagram packet which is then transmitted to the configured UDP server." - + " The user must ensure that the FlowFile content being fed to this processor is not larger than the maximum size for the underlying UDP transport. The maximum transport size will " - + "vary based on the platform setup but is generally just under 64KB. FlowFiles will be marked as failed if their content is larger than the maximum transport size.") +@CapabilityDescription("The PutUDP processor receives a FlowFile and packages the FlowFile content into a single UDP datagram packet which is then transmitted to the configured UDP server. " + + "A FlowFile larger than 65,507 bytes (the IPv4 UDP maximum payload) cannot be sent as one datagram and is routed to failure without reading content. " + + "The local UDP stack may still reject smaller datagrams; those FlowFiles are also marked as failed.") @InputRequirement(Requirement.INPUT_REQUIRED) @SeeAlso({ListenUDP.class, PutTCP.class}) @Tags({ "remote", "egress", "put", "udp" }) @SupportsBatching public class PutUDP extends AbstractPutEventProcessor<byte[]> { + /** + * Maximum UDP payload for IPv4: 65,535 byte IP packet minus 20 byte IPv4 header minus 8 byte UDP header. + */ + static final int MAX_IPV4_UDP_PAYLOAD_LENGTH = 65_535 - 20 - 8; + @Override public void onTrigger(final ProcessContext context, final ProcessSessionFactory sessionFactory) throws ProcessException { final ProcessSession session = sessionFactory.createSession(); @@ -55,6 +60,14 @@ public class PutUDP extends AbstractPutEventProcessor<byte[]> { return; } + if (flowFile.getSize() > MAX_IPV4_UDP_PAYLOAD_LENGTH) { + getLogger().error("Cannot send {} as a UDP datagram: size {} exceeds the IPv4 maximum payload of {} bytes", + flowFile, flowFile.getSize(), MAX_IPV4_UDP_PAYLOAD_LENGTH); + session.transfer(session.penalize(flowFile), REL_FAILURE); + session.commitAsync(); + return; + } + final StopWatch stopWatch = new StopWatch(true); try { final byte[] content = readContent(session, flowFile); diff --git a/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestPutUDP.java b/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestPutUDP.java index aee554d3498..d846823d434 100644 --- a/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestPutUDP.java +++ b/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestPutUDP.java @@ -39,6 +39,7 @@ import java.util.concurrent.LinkedBlockingQueue; import static org.junit.jupiter.api.Assertions.assertArrayEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; @Timeout(10) public class TestPutUDP { @@ -49,7 +50,7 @@ public class TestPutUDP { private static final Charset CHARSET = StandardCharsets.UTF_8; private static final int MAX_FRAME_LENGTH = 32800; private static final int VALID_LARGE_FILE_SIZE = 32768; - private static final int INVALID_LARGE_FILE_SIZE = 1_000_000; + private static final int OVERSIZE_UDP_PAYLOAD = PutUDP.MAX_IPV4_UDP_PAYLOAD_LENGTH + 1; private static final char CONTENT_CHAR = 'x'; private static final int DATA_WAIT_PERIOD = 50; private static final String[] EMPTY_FILE = {""}; @@ -100,13 +101,16 @@ public class TestPutUDP { } @Test - public void testSendLargeFileInvalid() throws Exception { + public void testSendLargerThanUdpPayloadLimit() throws Exception { configureProperties(); - String[] testData = createContent(INVALID_LARGE_FILE_SIZE); - sendMessages(testData); - checkRelationships(0, testData.length); + runner.enqueue(new byte[OVERSIZE_UDP_PAYLOAD]); + runner.run(); + + checkRelationships(0, 1); checkNoDataReceived(); runner.assertQueueEmpty(); + assertTrue(runner.getLogger().getErrorMessages().stream() + .anyMatch(message -> message.getMsg().contains("exceeds the IPv4 maximum payload"))); } @Test
