This is an automated email from the ASF dual-hosted git repository.
exceptionfactory pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new 56593ad NIFI-9546: Updated PutTCP and PutUDP with SupportsBatching
annotations
56593ad is described below
commit 56593ad12f40bbacd4627436eadc38a24f58ee16
Author: Mark Payne <[email protected]>
AuthorDate: Thu Jan 6 13:51:54 2022 -0500
NIFI-9546: Updated PutTCP and PutUDP with SupportsBatching annotations
- Removed incorrect usage of TriggerWhenEmpty
- Allow for 0 seconds of gracefully waiting for incoming threads on
shutdown since we won't have any
- Updated unit tests to no longer have arbitrary sleep statements
This closes #5639
Signed-off-by: David Handermann <[email protected]>
---
.../util/put/AbstractPutEventProcessor.java | 5 +-
.../apache/nifi/processors/standard/PutTCP.java | 92 ++---------------
.../apache/nifi/processors/standard/PutUDP.java | 113 ++++-----------------
.../nifi/processors/standard/TestPutTCP.java | 3 -
.../nifi/processors/standard/TestPutUDP.java | 26 ++---
5 files changed, 43 insertions(+), 196 deletions(-)
diff --git
a/nifi-nar-bundles/nifi-extension-utils/nifi-processor-utils/src/main/java/org/apache/nifi/processor/util/put/AbstractPutEventProcessor.java
b/nifi-nar-bundles/nifi-extension-utils/nifi-processor-utils/src/main/java/org/apache/nifi/processor/util/put/AbstractPutEventProcessor.java
index 41afeed..659e3df 100644
---
a/nifi-nar-bundles/nifi-extension-utils/nifi-processor-utils/src/main/java/org/apache/nifi/processor/util/put/AbstractPutEventProcessor.java
+++
b/nifi-nar-bundles/nifi-extension-utils/nifi-processor-utils/src/main/java/org/apache/nifi/processor/util/put/AbstractPutEventProcessor.java
@@ -193,7 +193,7 @@ public abstract class AbstractPutEventProcessor extends
AbstractSessionFactoryPr
* @return a list of relationships
*/
protected List<Relationship> getAdditionalRelationships() {
- return Collections.EMPTY_LIST;
+ return Collections.emptyList();
}
/**
@@ -202,7 +202,7 @@ public abstract class AbstractPutEventProcessor extends
AbstractSessionFactoryPr
* @return a list of properties
*/
protected List<PropertyDescriptor> getAdditionalProperties() {
- return Collections.EMPTY_LIST;
+ return Collections.emptyList();
}
@Override
@@ -251,6 +251,7 @@ public abstract class AbstractPutEventProcessor extends
AbstractSessionFactoryPr
factory.setMaxConnections(context.getMaxConcurrentTasks());
factory.setSocketSendBufferSize(context.getProperty(MAX_SOCKET_SEND_BUFFER_SIZE).asDataSize(DataUnit.B).intValue());
factory.setSingleEventPerConnection(singleEventPerConnection);
+ factory.setShutdownQuietPeriod(Duration.ZERO); // Quiet period not
necessary since sending threads will have completed before shutting down event
sender
final int timeout =
context.getProperty(TIMEOUT).evaluateAttributeExpressions().asTimePeriod(TimeUnit.MILLISECONDS).intValue();
factory.setTimeout(Duration.ofMillis(timeout));
diff --git
a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutTCP.java
b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutTCP.java
index da6e479..7bcdcc1 100644
---
a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutTCP.java
+++
b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutTCP.java
@@ -18,7 +18,7 @@ package org.apache.nifi.processors.standard;
import org.apache.nifi.annotation.behavior.InputRequirement;
import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
-import org.apache.nifi.annotation.behavior.TriggerWhenEmpty;
+import org.apache.nifi.annotation.behavior.SupportsBatching;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.annotation.documentation.SeeAlso;
import org.apache.nifi.annotation.documentation.Tags;
@@ -35,6 +35,7 @@ import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.io.InputStreamCallback;
import org.apache.nifi.processor.util.put.AbstractPutEventProcessor;
import org.apache.nifi.util.StopWatch;
+
import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.Charset;
@@ -42,44 +43,6 @@ import java.util.Arrays;
import java.util.List;
import java.util.concurrent.TimeUnit;
-/**
- * <p>
- * The PutTCP processor receives a FlowFile and transmits the FlowFile content
over a TCP connection to the configured TCP server. By default, the FlowFiles
are transmitted over the same TCP
- * connection (or pool of TCP connections if multiple input threads are
configured). To assist the TCP server with determining message boundaries, an
optional "Outgoing Message Delimiter" string can
- * be configured which is appended to the end of each FlowFiles content when
it is transmitted over the TCP connection. An optional "Connection Per
FlowFile" parameter can be specified to change the
- * behaviour so that each FlowFiles content is transmitted over a single TCP
connection which is opened when the FlowFile is received and closed after the
FlowFile has been sent. This option should
- * only be used for low message volume scenarios, otherwise the platform may
run out of TCP sockets.
- * </p>
- *
- * <p>
- * This processor has the following required properties:
- * <ul>
- * <li><b>Hostname</b> - The IP address or host name of the destination TCP
server.</li>
- * <li><b>Port</b> - The TCP port of the destination TCP server.</li>
- * </ul>
- * </p>
- *
- * <p>
- * This processor has the following optional properties:
- * <ul>
- * <li><b>Connection Per FlowFile</b> - Specifies that each FlowFiles content
will be transmitted on a separate TCP connection.</li>
- * <li><b>Idle Connection Expiration</b> - The time threshold after which a
TCP sender is deemed eligible for pruning - the associated TCP connection will
be closed after this timeout.</li>
- * <li><b>Max Size of Socket Send Buffer</b> - The maximum size of the socket
send buffer that should be used. This is a suggestion to the Operating System
to indicate how big the socket buffer should
- * be. If this value is set too low, the buffer may fill up before the data
can be read, and incoming data will be dropped.</li>
- * <li><b>Outgoing Message Delimiter</b> - A string to append to the end of
each FlowFiles content to indicate the end of the message to the TCP
server.</li>
- * <li><b>Timeout</b> - The timeout period for determining an error has
occurred whilst connecting or sending data.</li>
- * </ul>
- * </p>
- *
- * <p>
- * The following relationships are required:
- * <ul>
- * <li><b>failure</b> - Where to route FlowFiles that failed to be sent.</li>
- * <li><b>success</b> - Where to route FlowFiles after they were successfully
sent to the TCP server.</li>
- * </ul>
- * </p>
- *
- */
@CapabilityDescription("The PutTCP processor receives a FlowFile and transmits
the FlowFile content over a TCP connection to the configured TCP server. "
+ "By default, the FlowFiles are transmitted over the same TCP
connection (or pool of TCP connections if multiple input threads are
configured). "
+ "To assist the TCP server with determining message boundaries, an
optional \"Outgoing Message Delimiter\" string can be configured which is
appended "
@@ -87,20 +50,11 @@ import java.util.concurrent.TimeUnit;
+ "specified to change the behaviour so that each FlowFiles content is
transmitted over a single TCP connection which is opened when the FlowFile "
+ "is received and closed after the FlowFile has been sent. This
option should only be used for low message volume scenarios, otherwise the
platform " + "may run out of TCP sockets.")
@InputRequirement(Requirement.INPUT_REQUIRED)
-@SeeAlso(ListenTCP.class)
+@SeeAlso({ListenTCP.class, PutUDP.class})
@Tags({ "remote", "egress", "put", "tcp" })
-@TriggerWhenEmpty // trigger even when queue is empty so that the processor
can check for idle senders to prune.
+@SupportsBatching
public class PutTCP extends AbstractPutEventProcessor {
- /**
- * Creates a Universal Resource Identifier (URI) for this processor.
Constructs a URI of the form TCP://< host >:< port > where the host and port
- * values are taken from the configured property values.
- *
- * @param context
- * - the current process context.
- *
- * @return The URI value as a String.
- */
@Override
protected String createTransitUri(final ProcessContext context) {
final String protocol = TCP_VALUE.getValue();
@@ -110,11 +64,6 @@ public class PutTCP extends AbstractPutEventProcessor {
return new
StringBuilder().append(protocol).append("://").append(host).append(":").append(port).toString();
}
- /**
- * Get the additional properties that are used by this processor.
- *
- * @return List of PropertyDescriptors describing the additional
properties.
- */
@Override
protected List<PropertyDescriptor> getAdditionalProperties() {
return Arrays.asList(CONNECTION_PER_FLOWFILE,
@@ -124,16 +73,6 @@ public class PutTCP extends AbstractPutEventProcessor {
CHARSET);
}
- /**
- * event handler method to handle the FlowFile being forwarded to the
Processor by the framework. The FlowFile contents is sent out over a TCP
connection using an acquired ChannelSender object. If
- * the FlowFile contents was sent out successfully then the FlowFile is
forwarded to the success relationship. If an error occurred then the FlowFile
is forwarded to the failure relationship.
- *
- * @param context
- * - the current process context.
- *
- * @param sessionFactory
- * - a factory object to obtain a process session.
- */
@Override
public void onTrigger(final ProcessContext context, final
ProcessSessionFactory sessionFactory) throws ProcessException {
final ProcessSession session = sessionFactory.createSession();
@@ -161,30 +100,13 @@ public class PutTCP extends AbstractPutEventProcessor {
session.getProvenanceReporter().send(flowFile, transitUri,
stopWatch.getElapsed(TimeUnit.MILLISECONDS));
session.transfer(flowFile, REL_SUCCESS);
- session.commitAsync();
} catch (Exception e) {
- onFailure(context, session, flowFile);
- getLogger().error("Exception while handling a process session,
transferring {} to failure.", new Object[] { flowFile }, e);
+ getLogger().error("Exception while handling a process session,
transferring {} to failure.", flowFile, e);
+ session.transfer(session.penalize(flowFile), REL_FAILURE);
+ context.yield();
}
}
- /**
- * Event handler method to perform the required actions when a failure has
occurred. The FlowFile is penalized, forwarded to the failure relationship and
the context is yielded.
- *
- * @param context
- * - the current process context.
- *
- * @param session
- * - the current process session.
- * @param flowFile
- * - the FlowFile that has failed to have been processed.
- */
- protected void onFailure(final ProcessContext context, final
ProcessSession session, final FlowFile flowFile) {
- session.transfer(session.penalize(flowFile), REL_FAILURE);
- session.commitAsync();
- context.yield();
- }
-
@Override
protected String getProtocol(final ProcessContext context) {
return TCP_VALUE.getValue();
diff --git
a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutUDP.java
b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutUDP.java
index a6ce71c..9990701 100644
---
a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutUDP.java
+++
b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutUDP.java
@@ -18,7 +18,7 @@ package org.apache.nifi.processors.standard;
import org.apache.nifi.annotation.behavior.InputRequirement;
import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
-import org.apache.nifi.annotation.behavior.TriggerWhenEmpty;
+import org.apache.nifi.annotation.behavior.SupportsBatching;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.annotation.documentation.SeeAlso;
import org.apache.nifi.annotation.documentation.Tags;
@@ -27,7 +27,6 @@ import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.ProcessSessionFactory;
import org.apache.nifi.processor.exception.ProcessException;
-import org.apache.nifi.processor.io.InputStreamCallback;
import org.apache.nifi.processor.util.put.AbstractPutEventProcessor;
import org.apache.nifi.stream.io.StreamUtils;
import org.apache.nifi.util.StopWatch;
@@ -37,53 +36,19 @@ import java.io.IOException;
import java.io.InputStream;
import java.util.concurrent.TimeUnit;
-/**
- * <p>
- * 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.
- * </p>
- *
- * <p>
- * This processor has the following required properties:
- * <ul>
- * <li><b>Hostname</b> - The IP address or host name of the destination UDP
server.</li>
- * <li><b>Port</b> - The UDP port of the destination UDP server.</li>
- * </ul>
- * </p>
- *
- * <p>
- * This processor has the following optional properties:
- * <ul>
- * <li><b>Max Size of Socket Send Buffer</b> - The maximum size of the socket
send buffer that should be used. This is a suggestion to the Operating System
to indicate how big the socket buffer should
- * be. If this value is set too low, the buffer may fill up before the data
can be read, and incoming data will be dropped.</li>
- * <li><b>Idle Connection Expiration</b> - The time threshold after which a
UDP Datagram sender is deemed eligible for pruning.</li>
- * </ul>
- * </p>
- *
- * <p>
- * The following relationships are required:
- * <ul>
- * <li><b>failure</b> - Where to route FlowFiles that failed to be sent.</li>
- * <li><b>success</b> - Where to route FlowFiles after they were successfully
sent to the UDP server.</li>
- * </ul>
- * </p>
- *
- */
@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.")
@InputRequirement(Requirement.INPUT_REQUIRED)
-@SeeAlso(ListenUDP.class)
+@SeeAlso({ListenUDP.class, PutTCP.class})
@Tags({ "remote", "egress", "put", "udp" })
-@TriggerWhenEmpty // trigger even when queue is empty so that the processor
can check for idle senders to prune.
+@SupportsBatching
public class PutUDP extends AbstractPutEventProcessor {
/**
* Creates a Universal Resource Identifier (URI) for this processor.
Constructs a URI of the form UDP://host:port where the host and port values are
taken from the configured property values.
*
- * @param context
- * - the current process context.
+ * @param context - the current process context.
*
* @return The URI value as a String.
*/
@@ -93,19 +58,9 @@ public class PutUDP extends AbstractPutEventProcessor {
final String host =
context.getProperty(HOSTNAME).evaluateAttributeExpressions().getValue();
final String port =
context.getProperty(PORT).evaluateAttributeExpressions().getValue();
- return new
StringBuilder().append(protocol).append("://").append(host).append(":").append(port).toString();
+ return protocol + "://" + host + ":" + port;
}
- /**
- * event handler method to handle the FlowFile being forwarded to the
Processor by the framework. The FlowFile contents is sent out as a UDP datagram
using an acquired ChannelSender object. If the
- * FlowFile contents was sent out successfully then the FlowFile is
forwarded to the success relationship. If an error occurred then the FlowFile
is forwarded to the failure relationship.
- *
- * @param context
- * - the current process context.
- *
- * @param sessionFactory
- * - a factory object to obtain a process session.
- */
@Override
public void onTrigger(final ProcessContext context, final
ProcessSessionFactory sessionFactory) throws ProcessException {
final ProcessSession session = sessionFactory.createSession();
@@ -115,61 +70,31 @@ public class PutUDP extends AbstractPutEventProcessor {
}
try {
- final byte[] content = readContent(session, flowFile);
StopWatch stopWatch = new StopWatch(true);
- if (content != null) {
- eventSender.sendEvent(content);
- }
+ final byte[] content = readContent(session, flowFile);
+ eventSender.sendEvent(content);
+
session.getProvenanceReporter().send(flowFile, transitUri,
stopWatch.getElapsed(TimeUnit.MILLISECONDS));
session.transfer(flowFile, REL_SUCCESS);
session.commitAsync();
} catch (Exception e) {
- getLogger().error("Exception while handling a process session,
transferring {} to failure.", new Object[] { flowFile }, e);
- onFailure(context, session, flowFile);
+ getLogger().error("Exception while handling a process session,
transferring {} to failure.", new Object[]{flowFile}, e);
+ session.transfer(session.penalize(flowFile), REL_FAILURE);
+ context.yield();
}
}
- /**
- * event handler method to perform the required actions when a failure has
occurred. The FlowFile is penalized, forwarded to the failure relationship and
the context is yielded.
- *
- * @param context
- * - the current process context.
- *
- * @param session
- * - the current process session.
- * @param flowFile
- * - the FlowFile that has failed to have been processed.
- */
- protected void onFailure(final ProcessContext context, final
ProcessSession session, final FlowFile flowFile) {
- session.transfer(session.penalize(flowFile), REL_FAILURE);
- session.commitAsync();
- context.yield();
+ @Override
+ protected String getProtocol(final ProcessContext context) {
+ return UDP_VALUE.getValue();
}
- /**
- * Helper method to read the FlowFile content stream into a byte array.
- *
- * @param session
- * - the current process session.
- * @param flowFile
- * - the FlowFile to read the content from.
- *
- * @return byte array representation of the FlowFile content.
- */
- protected byte[] readContent(final ProcessSession session, final FlowFile
flowFile) {
- final ByteArrayOutputStream baos = new ByteArrayOutputStream((int)
flowFile.getSize() + 1);
- session.read(flowFile, new InputStreamCallback() {
- @Override
- public void process(final InputStream in) throws IOException {
- StreamUtils.copy(in, baos);
- }
- });
+ private byte[] readContent(final ProcessSession session, final FlowFile
flowFile) throws IOException {
+ final ByteArrayOutputStream baos = new ByteArrayOutputStream((int)
flowFile.getSize());
+ try (final InputStream in = session.read(flowFile)) {
+ StreamUtils.copy(in, baos);
+ }
return baos.toByteArray();
}
-
- @Override
- protected String getProtocol(final ProcessContext context) {
- return UDP_VALUE.getValue();
- }
}
diff --git
a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestPutTCP.java
b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestPutTCP.java
index baad1e8..549c3a0 100644
---
a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestPutTCP.java
+++
b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestPutTCP.java
@@ -149,7 +149,6 @@ public class TestPutTCP {
assertMessagesReceived(VALID_FILES);
runner.setProperty(PutTCP.IDLE_EXPIRATION, "500 ms");
- Thread.sleep(1000);
runner.run(1, false, false);
runner.clearTransferState();
sendTestData(VALID_FILES);
@@ -181,7 +180,6 @@ public class TestPutTCP {
shutdownServer();
sendTestData(VALID_FILES);
- Thread.sleep(500);
runner.assertQueueEmpty();
configureProperties(TCP_SERVER_ADDRESS, OUTGOING_MESSAGE_DELIMITER,
false);
@@ -212,7 +210,6 @@ public class TestPutTCP {
public void testRunSuccessFiveHundredMessages() throws Exception {
configureProperties(TCP_SERVER_ADDRESS, OUTGOING_MESSAGE_DELIMITER,
false);
createTestServer(port);
- Thread.sleep(1000);
final String[] testData = createContent(VALID_SMALL_FILE_SIZE);
sendTestData(testData, LOAD_TEST_ITERATIONS, LOAD_TEST_THREAD_COUNT);
assertMessagesReceived(testData, LOAD_TEST_ITERATIONS);
diff --git
a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestPutUDP.java
b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestPutUDP.java
index 4884e7d..104cec1 100644
---
a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestPutUDP.java
+++
b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestPutUDP.java
@@ -16,16 +16,6 @@
*/
package org.apache.nifi.processors.standard;
-import static org.junit.Assert.assertArrayEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNull;
-
-import java.net.InetAddress;
-import java.nio.charset.Charset;
-import java.nio.charset.StandardCharsets;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.LinkedBlockingQueue;
-
import org.apache.nifi.event.transport.EventServer;
import org.apache.nifi.event.transport.configuration.ShutdownQuietPeriod;
import org.apache.nifi.event.transport.configuration.ShutdownTimeout;
@@ -40,6 +30,16 @@ import org.junit.After;
import org.junit.Before;
import org.junit.Test;
+import java.net.InetAddress;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+
public class TestPutUDP {
private final static String UDP_SERVER_ADDRESS = "127.0.0.1";
@@ -50,13 +50,13 @@ public class TestPutUDP {
private final static int MAX_FRAME_LENGTH = 32800;
private final static int VALID_LARGE_FILE_SIZE = 32768;
private final static int VALID_SMALL_FILE_SIZE = 64;
- private final static int INVALID_LARGE_FILE_SIZE = 1000000;
+ private final static int INVALID_LARGE_FILE_SIZE = 1_000_000;
private final static int LOAD_TEST_ITERATIONS = 500;
private final static int LOAD_TEST_THREAD_COUNT = 1;
private final static int DEFAULT_ITERATIONS = 1;
private final static int DEFAULT_THREAD_COUNT = 1;
private final static char CONTENT_CHAR = 'x';
- private final static int DATA_WAIT_PERIOD = 1000;
+ private final static int DATA_WAIT_PERIOD = 50;
private final static int DEFAULT_TEST_TIMEOUT_PERIOD = 10000;
private final static int LONG_TEST_TIMEOUT_PERIOD = 100000;
@@ -153,10 +153,12 @@ public class TestPutUDP {
sendTestData(VALID_FILES);
checkReceivedAllData(VALID_FILES);
reset(port);
+
configureProperties(UDP_SERVER_ADDRESS);
sendTestData(VALID_FILES);
checkReceivedAllData(VALID_FILES);
reset(port);
+
configureProperties(UDP_SERVER_ADDRESS);
sendTestData(VALID_FILES);
checkReceivedAllData(VALID_FILES);
