null if none is configured
+ */
+ String getKeystoreFilename();
+
+ /**
+ * @return the password to use for the keystore, or null if none is configured
+ */
+ String getKeystorePassword();
+
+ /**
+ * @return the Type of the keystore, or null if none is configured
+ */
+ KeystoreType getKeystoreType();
+
+ /**
+ * @return the filename to use for the truststore, or null if none is configured
+ */
+ String getTruststoreFilename();
+
+ /**
+ * @return the password to use for the truststore, or null if none is configured
+ */
+ String getTruststorePassword();
+
+ /**
+ * @return the type of the truststore, or null if none is configured
+ */
+ KeystoreType getTruststoreType();
+
/**
* @return the file that is to be used for persisting the nodes of a remote
- * cluster, if any
+ * cluster, if any
*/
File getPeerPersistenceFile();
diff --git a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/client/socket/EndpointConnection.java b/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/client/socket/EndpointConnection.java
similarity index 100%
rename from nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/client/socket/EndpointConnection.java
rename to nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/client/socket/EndpointConnection.java
diff --git a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/client/socket/EndpointConnectionPool.java b/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/client/socket/EndpointConnectionPool.java
similarity index 100%
rename from nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/client/socket/EndpointConnectionPool.java
rename to nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/client/socket/EndpointConnectionPool.java
diff --git a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/client/socket/SocketClient.java b/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/client/socket/SocketClient.java
similarity index 100%
rename from nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/client/socket/SocketClient.java
rename to nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/client/socket/SocketClient.java
diff --git a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/cluster/AdaptedNodeInformation.java b/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/cluster/AdaptedNodeInformation.java
similarity index 100%
rename from nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/cluster/AdaptedNodeInformation.java
rename to nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/cluster/AdaptedNodeInformation.java
diff --git a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/cluster/ClusterNodeInformation.java b/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/cluster/ClusterNodeInformation.java
similarity index 100%
rename from nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/cluster/ClusterNodeInformation.java
rename to nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/cluster/ClusterNodeInformation.java
diff --git a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/cluster/NodeInformation.java b/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/cluster/NodeInformation.java
similarity index 100%
rename from nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/cluster/NodeInformation.java
rename to nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/cluster/NodeInformation.java
diff --git a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/cluster/NodeInformationAdapter.java b/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/cluster/NodeInformationAdapter.java
similarity index 100%
rename from nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/cluster/NodeInformationAdapter.java
rename to nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/cluster/NodeInformationAdapter.java
diff --git a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/codec/FlowFileCodec.java b/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/codec/FlowFileCodec.java
similarity index 100%
rename from nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/codec/FlowFileCodec.java
rename to nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/codec/FlowFileCodec.java
diff --git a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/codec/StandardFlowFileCodec.java b/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/codec/StandardFlowFileCodec.java
similarity index 100%
rename from nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/codec/StandardFlowFileCodec.java
rename to nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/codec/StandardFlowFileCodec.java
diff --git a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/exception/HandshakeException.java b/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/exception/HandshakeException.java
similarity index 100%
rename from nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/exception/HandshakeException.java
rename to nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/exception/HandshakeException.java
diff --git a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/exception/PortNotRunningException.java b/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/exception/PortNotRunningException.java
similarity index 100%
rename from nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/exception/PortNotRunningException.java
rename to nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/exception/PortNotRunningException.java
diff --git a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/exception/ProtocolException.java b/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/exception/ProtocolException.java
similarity index 100%
rename from nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/exception/ProtocolException.java
rename to nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/exception/ProtocolException.java
diff --git a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/exception/UnknownPortException.java b/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/exception/UnknownPortException.java
similarity index 100%
rename from nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/exception/UnknownPortException.java
rename to nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/exception/UnknownPortException.java
diff --git a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/io/socket/SocketChannelCommunicationsSession.java b/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/io/socket/SocketChannelCommunicationsSession.java
similarity index 100%
rename from nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/io/socket/SocketChannelCommunicationsSession.java
rename to nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/io/socket/SocketChannelCommunicationsSession.java
diff --git a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/io/socket/SocketChannelInput.java b/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/io/socket/SocketChannelInput.java
similarity index 100%
rename from nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/io/socket/SocketChannelInput.java
rename to nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/io/socket/SocketChannelInput.java
diff --git a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/io/socket/SocketChannelOutput.java b/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/io/socket/SocketChannelOutput.java
similarity index 100%
rename from nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/io/socket/SocketChannelOutput.java
rename to nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/io/socket/SocketChannelOutput.java
diff --git a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/io/socket/ssl/SSLSocketChannelCommunicationsSession.java b/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/io/socket/ssl/SSLSocketChannelCommunicationsSession.java
similarity index 100%
rename from nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/io/socket/ssl/SSLSocketChannelCommunicationsSession.java
rename to nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/io/socket/ssl/SSLSocketChannelCommunicationsSession.java
diff --git a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/io/socket/ssl/SSLSocketChannelInput.java b/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/io/socket/ssl/SSLSocketChannelInput.java
similarity index 100%
rename from nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/io/socket/ssl/SSLSocketChannelInput.java
rename to nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/io/socket/ssl/SSLSocketChannelInput.java
diff --git a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/io/socket/ssl/SSLSocketChannelOutput.java b/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/io/socket/ssl/SSLSocketChannelOutput.java
similarity index 100%
rename from nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/io/socket/ssl/SSLSocketChannelOutput.java
rename to nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/io/socket/ssl/SSLSocketChannelOutput.java
diff --git a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/ClientProtocol.java b/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/ClientProtocol.java
similarity index 100%
rename from nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/ClientProtocol.java
rename to nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/ClientProtocol.java
diff --git a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/CommunicationsInput.java b/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/CommunicationsInput.java
similarity index 100%
rename from nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/CommunicationsInput.java
rename to nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/CommunicationsInput.java
diff --git a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/CommunicationsOutput.java b/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/CommunicationsOutput.java
similarity index 100%
rename from nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/CommunicationsOutput.java
rename to nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/CommunicationsOutput.java
diff --git a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/CommunicationsSession.java b/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/CommunicationsSession.java
similarity index 100%
rename from nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/CommunicationsSession.java
rename to nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/CommunicationsSession.java
diff --git a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/DataPacket.java b/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/DataPacket.java
similarity index 100%
rename from nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/DataPacket.java
rename to nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/DataPacket.java
diff --git a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/RequestType.java b/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/RequestType.java
similarity index 100%
rename from nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/RequestType.java
rename to nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/RequestType.java
diff --git a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/HandshakeProperty.java b/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/HandshakeProperty.java
similarity index 100%
rename from nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/HandshakeProperty.java
rename to nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/HandshakeProperty.java
diff --git a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/Response.java b/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/Response.java
similarity index 100%
rename from nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/Response.java
rename to nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/Response.java
diff --git a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/ResponseCode.java b/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/ResponseCode.java
similarity index 100%
rename from nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/ResponseCode.java
rename to nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/ResponseCode.java
diff --git a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/SocketClientProtocol.java b/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/SocketClientProtocol.java
similarity index 100%
rename from nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/SocketClientProtocol.java
rename to nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/SocketClientProtocol.java
diff --git a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/SocketClientTransaction.java b/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/SocketClientTransaction.java
similarity index 100%
rename from nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/SocketClientTransaction.java
rename to nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/SocketClientTransaction.java
diff --git a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/SocketClientTransactionCompletion.java b/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/SocketClientTransactionCompletion.java
similarity index 100%
rename from nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/SocketClientTransactionCompletion.java
rename to nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/SocketClientTransactionCompletion.java
diff --git a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/util/NiFiRestApiUtil.java b/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/util/NiFiRestApiUtil.java
similarity index 100%
rename from nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/util/NiFiRestApiUtil.java
rename to nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/util/NiFiRestApiUtil.java
diff --git a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/util/PeerStatusCache.java b/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/util/PeerStatusCache.java
similarity index 100%
rename from nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/util/PeerStatusCache.java
rename to nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/util/PeerStatusCache.java
diff --git a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/util/StandardDataPacket.java b/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/util/StandardDataPacket.java
similarity index 100%
rename from nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/util/StandardDataPacket.java
rename to nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/util/StandardDataPacket.java
diff --git a/nifi/nifi-commons/nifi-site-to-site-client/src/test/java/org/apache/nifi/remote/client/socket/TestEndpointConnectionStatePool.java b/nifi-commons/nifi-site-to-site-client/src/test/java/org/apache/nifi/remote/client/socket/TestEndpointConnectionStatePool.java
similarity index 100%
rename from nifi/nifi-commons/nifi-site-to-site-client/src/test/java/org/apache/nifi/remote/client/socket/TestEndpointConnectionStatePool.java
rename to nifi-commons/nifi-site-to-site-client/src/test/java/org/apache/nifi/remote/client/socket/TestEndpointConnectionStatePool.java
diff --git a/nifi/nifi-commons/nifi-site-to-site-client/src/test/java/org/apache/nifi/remote/client/socket/TestSiteToSiteClient.java b/nifi-commons/nifi-site-to-site-client/src/test/java/org/apache/nifi/remote/client/socket/TestSiteToSiteClient.java
similarity index 100%
rename from nifi/nifi-commons/nifi-site-to-site-client/src/test/java/org/apache/nifi/remote/client/socket/TestSiteToSiteClient.java
rename to nifi-commons/nifi-site-to-site-client/src/test/java/org/apache/nifi/remote/client/socket/TestSiteToSiteClient.java
diff --git a/nifi/nifi-commons/nifi-socket-utils/pom.xml b/nifi-commons/nifi-socket-utils/pom.xml
similarity index 98%
rename from nifi/nifi-commons/nifi-socket-utils/pom.xml
rename to nifi-commons/nifi-socket-utils/pom.xml
index de1aa0994e..4ed05ef99f 100644
--- a/nifi/nifi-commons/nifi-socket-utils/pom.xml
+++ b/nifi-commons/nifi-socket-utils/pom.xml
@@ -18,7 +18,7 @@
+ This processor executes an Apache Flume sink. FlowFiles are wrapped in Flume's + Event interface. The content of the FlowFile becomes the body of the Event and + the attributes of the FlowFile become Event headers. The following special + headers are also set: +
+| Flume Event Header | +FlowFile Attribute | +
|---|---|
| nifi.entry.date | +FlowFile#getEntryDate() | +
| nifi.id | +FlowFile#getId() | +
| nifi.last.queue.date | +FlowFile#getLastQueueDate() | +
| nifi.lineage.identifiers.${i} | +FlowFile#getLineageIdentifiers()[i] | +
| nifi.lineage.start.date | +FlowFile#getLineageStartDate() | +
| nifi.size | +FlowFile#getSize() | +
+ In NiFi, the contents of a FlowFile are accessed via a stream, but in Flume it is + stored in a byte array. This means the full content will be loaded into memory when + a FlowFile is processed by the ExecuteFlumeSink processor. You should consider the + typical size of the FlowFiles you'll process and the batch size, if any, your sink + is configured with when setting NiFi's heap size. +
++ This processor is designed to execute arbitrary Flume sinks. Most of the details + of configuring the sink is deferred to Flume's built-in configuration system. + For details on the available settings for each sink type, refer to the Flume + User Guide. + Configuring the Flume sink is a four step process: +
+tier1
+ tier1, then the Sink Name
+ is the value of the tier1.sinks property. Example: sink-1
+ tier1.sinks.sink-1.
+ Do not copy the tier1.sinks.sink-1.type or
+ tier1.sinks.sink-1.channel properties.
+ + Assuming you had the following existing Flume configuration file: +
++a1.sources = r1 +a1.sinks = k1 +a1.channels = c1 + +a1.sources.r1.type = netcat +a1.sources.r1.bind = localhost +a1.sources.r1.port = 44444 + +a1.sinks.k1.type = hdfs +a1.sinks.k1.channel = c1 +a1.sinks.k1.hdfs.path = /flume/events/%y-%m-%d/%H%M/%S +a1.sinks.k1.hdfs.filePrefix = events- +a1.sinks.k1.hdfs.round = true +a1.sinks.k1.hdfs.roundValue = 10 +a1.sinks.k1.hdfs.roundUnit = minute + +a1.channels.c1.type = memory +a1.channels.c1.capacity = 1000 +a1.channels.c1.transactionCapacity = 100 + +a1.sources.r1.channels = c1 +a1.sinks.k1.channel = c1+
+ Then you'd configure the ExecuteFlumeSink as follows: +
+| Property | +Value | +
|---|---|
| Sink Type | +hdfs | +
| Agent Name | +a1 | +
| Sink Name | +k1 | +
| Flume Configuration | +
+
+ a1.sinks.k1.hdfs.path = /flume/events/%y-%m-%d/%H%M/%S
+ |
+
+ This processor executes an Apache Flume source. Each Flume Event is turned into + a FlowFile. The content of the FlowFile is set to the body of the Event and + the Event headers become FlowFile attributes. +
++ This processor is designed to execute arbitrary Flume sources. Most of the details + of configuring the source is deferred to Flume's built-in configuration system. + For details on the available settings for each source type, refer to the Flume + User Guide. + Configuring the Flume source is a four step process: +
+tier1
+ tier1, then the Source Name
+ is the value of the tier1.sources property. Example: src-1
+ tier1.sources.src-1.
+ Do not copy the tier1.sources.src-1.type or
+ tier1.sources.src-1.channel properties.
+ + Assuming you had the following existing Flume configuration file: +
++a1.sources = r1 +a1.sinks = k1 +a1.channels = c1 + +a1.sources.r1.type = multiport_syslogtcp +a1.sources.r1.channels = c1 +a1.sources.r1.host = 0.0.0.0 +a1.sources.r1.ports = 10001 10002 10003 +a1.sources.r1.portHeader = port + +a1.sinks.k1.type = logger + +a1.channels.c1.type = memory +a1.channels.c1.capacity = 1000 +a1.channels.c1.transactionCapacity = 100 + +a1.sources.r1.channels = c1 +a1.sinks.k1.channel = c1+
+ Then you'd configure the ExecuteFlumeSource as follows: +
+| Property | +Value | +
|---|---|
| Source Type | +multiport_syslogtcp | +
| Agent Name | +a1 | +
| Source Name | +r1 | +
| Flume Configuration | +
+
+ a1.sources.r1.host = 0.0.0.0
+ |
+
This processor is used to convert data between two Avro formats, such as those coming from the ConvertCSVToAvro or
+ ConvertJSONToAvro processors. The input and output content of the flow files should be Avro data files. The processor
+ includes support for the following basic type conversions:
+
+ Throughout this example, we will refer to input data with the following schema: +
+{
+ "type": "record",
+ "name": "CustomerInput",
+ "namespace": "org.apache.example",
+ "fields": [
+ {
+ "name": "id",
+ "type": "string"
+ },
+ {
+ "name": "companyName",
+ "type": ["null", "string"],
+ "default": null
+ },
+ {
+ "name": "revenue",
+ "type": ["null", "string"],
+ "default": null
+ },
+ {
+ "name" : "parent",
+ "type" : [ "null", {
+ "type" : "record",
+ "name" : "parent",
+ "fields" : [ {
+ "name" : "name",
+ "type" : ["null", "string"],
+ "default" : null
+ }, {
+ "name" : "id",
+ "type" : "string"
+ } ]
+ } ],
+ "default" : null
+ }
+ ]
+}
+
+ Where even though the revenue and id fields are mapped as string, they are logically long and double respectively.
+ By default, fields with matching names will be mapped automatically, so the following output schema could be converted
+ without using dynamic properties:
+
+{
+ "type": "record",
+ "name": "SimpleCustomerOutput",
+ "namespace": "org.apache.example",
+ "fields": [
+ {
+ "name": "id",
+ "type": "long"
+ },
+ {
+ "name": "companyName",
+ "type": ["null", "string"],
+ "default": null
+ },
+ {
+ "name": "revenue",
+ "type": ["null", "double"],
+ "default": null
+ }
+ ]
+}
+
+ To rename companyName to name and to extract the parent's id field, both a schema and a dynamic properties must be provided.
+ For example, to convert to the following schema:
+
+{
+ "type": "record",
+ "name": "SimpleCustomerOutput",
+ "namespace": "org.apache.example",
+ "fields": [
+ {
+ "name": "id",
+ "type": "long"
+ },
+ {
+ "name": "name",
+ "type": ["null", "string"],
+ "default": null
+ },
+ {
+ "name": "revenue",
+ "type": ["null", "double"],
+ "default": null
+ },
+ {
+ "name": "parentId",
+ "type": ["null", "long"],
+ "default": null
+ }
+ ]
+}
+
+ The following dynamic properties would be used:
+ +"companyName" -> "name" +"parent.id" -> "parentId" ++ + + diff --git a/nifi-nar-bundles/nifi-kite-bundle/nifi-kite-processors/src/test/java/org/apache/nifi/processors/kite/TestAvroRecordConverter.java b/nifi-nar-bundles/nifi-kite-bundle/nifi-kite-processors/src/test/java/org/apache/nifi/processors/kite/TestAvroRecordConverter.java new file mode 100644 index 0000000000..1a4748f1b6 --- /dev/null +++ b/nifi-nar-bundles/nifi-kite-bundle/nifi-kite-processors/src/test/java/org/apache/nifi/processors/kite/TestAvroRecordConverter.java @@ -0,0 +1,201 @@ +/* + * 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.processors.kite; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; + +import java.util.Collections; +import java.util.Map; + +import org.apache.avro.Schema; +import org.apache.avro.SchemaBuilder; +import org.apache.avro.generic.GenericData.Record; +import org.codehaus.jackson.map.ObjectMapper; +import org.junit.Test; + +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; + +public class TestAvroRecordConverter { + final static ObjectMapper OBJECT_MAPPER = new ObjectMapper(); + final static Map
+ * MUST be called with the write lock held. + *
+ * + * Rolls over the data in the journal files, merging them into a single Provenance Event Log File, and + * compressing and indexing as needed. * * @param force if true, will force a rollover regardless of whether or not data has been written * @throws IOException if unable to complete rollover @@ -968,7 +989,7 @@ public class PersistentProvenanceRepository implements ProvenanceEventRepository // If this is the first time we're creating the out stream, or if we // have written something to the stream, then roll over - if (recordsWrittenSinceRollover.get() > 0L || repoDirty.get() || force) { + if (force || recordsWrittenSinceRollover.get() > 0L || dirtyWriterCount.get() > 0) { final List+ * Merges all of the given Journal Files into a single, merged Provenance Event Log File. As these records are merged, they will be compressed, if the repository is configured to compress records, + * and will be indexed. + *
+ * + *
+ * If the repository is configured to compress the data, the file written to may not be the same as the suggestedMergeFile, as a filename extension of '.gz' may be appended. If the
+ * journals are successfully merged, the file that they were merged into will be returned. If unable to merge the records (for instance, because the repository has been closed or because the list
+ * of journal files was empty), this method will return null.
+ *
null if no records were merged.
+ *
+ * @throws IOException if a problem occurs writing to the mergedFile, reading from a journal, or updating the Lucene Index.
+ */
+ File mergeJournals(final Listfalse; otherwise, obtains
- * the lock and returns true.
+ * not immediately available, or if the writer is 'dirty' (see {@link #markDirty()},
+ * returns false; otherwise, obtains the lock and returns true.
*
* @return true if the lock was obtained, false otherwise.
*/
boolean tryLock();
+ /**
+ * Indicates that this Record Writer is 'dirty', meaning that it can no longer be
+ * updated. This can happen, for example, if a partial record is written. In this case,
+ * writing to this RecordWriter again could cause corruption.
+ */
+ void markDirty();
+
/**
* Syncs the content written to this writer to disk.
+ *
* @throws IOException if unable to sync content to disk
*/
void sync() throws IOException;
diff --git a/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/serialization/RecordWriters.java b/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/serialization/RecordWriters.java
similarity index 100%
rename from nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/serialization/RecordWriters.java
rename to nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/serialization/RecordWriters.java
diff --git a/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/toc/StandardTocReader.java b/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/toc/StandardTocReader.java
similarity index 100%
rename from nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/toc/StandardTocReader.java
rename to nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/toc/StandardTocReader.java
diff --git a/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/toc/StandardTocWriter.java b/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/toc/StandardTocWriter.java
similarity index 100%
rename from nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/toc/StandardTocWriter.java
rename to nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/toc/StandardTocWriter.java
diff --git a/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/toc/TocReader.java b/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/toc/TocReader.java
similarity index 100%
rename from nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/toc/TocReader.java
rename to nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/toc/TocReader.java
diff --git a/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/toc/TocUtil.java b/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/toc/TocUtil.java
similarity index 100%
rename from nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/toc/TocUtil.java
rename to nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/toc/TocUtil.java
diff --git a/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/toc/TocWriter.java b/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/toc/TocWriter.java
similarity index 100%
rename from nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/toc/TocWriter.java
rename to nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/toc/TocWriter.java
diff --git a/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/resources/META-INF/services/org.apache.nifi.provenance.ProvenanceEventRepository b/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/resources/META-INF/services/org.apache.nifi.provenance.ProvenanceEventRepository
similarity index 100%
rename from nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/resources/META-INF/services/org.apache.nifi.provenance.ProvenanceEventRepository
rename to nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/resources/META-INF/services/org.apache.nifi.provenance.ProvenanceEventRepository
diff --git a/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/test/java/org/apache/nifi/provenance/TestPersistentProvenanceRepository.java b/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/test/java/org/apache/nifi/provenance/TestPersistentProvenanceRepository.java
similarity index 90%
rename from nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/test/java/org/apache/nifi/provenance/TestPersistentProvenanceRepository.java
rename to nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/test/java/org/apache/nifi/provenance/TestPersistentProvenanceRepository.java
index 16f0312839..713180f206 100644
--- a/nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/test/java/org/apache/nifi/provenance/TestPersistentProvenanceRepository.java
+++ b/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/test/java/org/apache/nifi/provenance/TestPersistentProvenanceRepository.java
@@ -28,10 +28,12 @@ import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
@@ -58,15 +60,21 @@ import org.apache.nifi.provenance.search.SearchTerms;
import org.apache.nifi.provenance.search.SearchableField;
import org.apache.nifi.provenance.serialization.RecordReader;
import org.apache.nifi.provenance.serialization.RecordReaders;
+import org.apache.nifi.provenance.serialization.RecordWriter;
import org.apache.nifi.reporting.Severity;
+import org.apache.nifi.stream.io.DataOutputStream;
import org.apache.nifi.util.file.FileUtils;
import org.junit.After;
+import org.junit.Assert;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Ignore;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TestName;
+import org.mockito.Mockito;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
public class TestPersistentProvenanceRepository {
@@ -135,6 +143,8 @@ public class TestPersistentProvenanceRepository {
private EventReporter getEventReporter() {
return new EventReporter() {
+ private static final long serialVersionUID = 1L;
+
@Override
public void reportEvent(Severity severity, String category, String message) {
System.out.println(severity + " : " + category + " : " + message);
@@ -547,7 +557,12 @@ public class TestPersistentProvenanceRepository {
builder.fromFlowFile(createFlowFile(3L, 3000L, attributes));
for (int i = 0; i < 10; i++) {
- attributes.put("uuid", "00000000-0000-0000-0000-00000000000" + String.valueOf(i + j * 10));
+ String uuidSuffix = String.valueOf(i + j * 10);
+ if (uuidSuffix.length() < 2) {
+ uuidSuffix = "0" + uuidSuffix;
+ }
+
+ attributes.put("uuid", "00000000-0000-0000-0000-0000000000" + uuidSuffix);
builder.fromFlowFile(createFlowFile(i + j * 10, 3000L, attributes));
repo.registerEvent(builder.build());
}
@@ -1200,4 +1215,109 @@ public class TestPersistentProvenanceRepository {
assertEquals("12345678901234567890123456789012345678901234567890", retrieved.getAttributes().get("75chars"));
}
+ @Test
+ public void testBehaviorOnOutOfMemory() throws IOException, InterruptedException {
+ final RepositoryConfiguration config = createConfiguration();
+ config.setMaxEventFileLife(3, TimeUnit.MINUTES);
+ config.setJournalCount(4);
+
+ // Create a repository that overrides the createWriters() method so that we can return writers that will throw
+ // OutOfMemoryError where we want to
+ final AtomicBoolean causeOOME = new AtomicBoolean(false);
+ repo = new PersistentProvenanceRepository(config, DEFAULT_ROLLOVER_MILLIS) {
+ @Override
+ protected RecordWriter[] createWriters(RepositoryConfiguration config, long initialRecordId) throws IOException {
+ final RecordWriter[] recordWriters = super.createWriters(config, initialRecordId);
+
+ // Spy on each of the writers so that a call to writeUUID throws an OutOfMemoryError if we set the
+ // causeOOME flag to true
+ final StandardRecordWriter[] spiedWriters = new StandardRecordWriter[recordWriters.length];
+ for (int i = 0; i < recordWriters.length; i++) {
+ final StandardRecordWriter writer = (StandardRecordWriter) recordWriters[i];
+
+ spiedWriters[i] = Mockito.spy(writer);
+ Mockito.doAnswer(new Answer