Apache
/
nifi
mirror of https://github.com/apache/nifi.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

NIFI-11179 Removed Flume components 

This closes #6950

Signed-off-by: David Handermann <exceptionfactory@apache.org>
This commit is contained in:
Pierre Villard 2023-02-14 15:16:57 -05:00 committed by exceptionfactory
parent 0d0ba1ca41
commit 35cf9f87be
No known key found for this signature in database
GPG Key ID: 29B6A52D2AAE8DBA
26 changed files with 0 additions and 2621 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

136
nifi-nar-bundles/nifi-flume-bundle/nifi-flume-nar/pom.xml
View File

@ -1,136 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
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.
-->
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-flume-bundle</artifactId>
<version>2.0.0-SNAPSHOT</version>
</parent>
<artifactId>nifi-flume-nar</artifactId>
<packaging>nar</packaging>
<properties>
<maven.javadoc.skip>true</maven.javadoc.skip>
<source.skip>true</source.skip>
</properties>
<dependencies>
<dependency>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-flume-processors</artifactId>
<!-- The following are inherited from nifi-hadoop-libraries-nar -->
<exclusions>
<exclusion>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-client</artifactId>
</exclusion>
<exclusion>
<groupId>org.xerial.snappy</groupId>
<artifactId>snappy-java</artifactId>
</exclusion>
<exclusion>
<groupId>com.thoughtworks.paranamer</groupId>
<artifactId>paranamer</artifactId>
</exclusion>
<exclusion>
<groupId>org.codehaus.jackson</groupId>
<artifactId>jackson-mapper-asl</artifactId>
</exclusion>
<exclusion>
<groupId>org.codehaus.jackson</groupId>
<artifactId>jackson-core-asl</artifactId>
</exclusion>
<exclusion>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
</exclusion>
<exclusion>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
</exclusion>
<exclusion>
<groupId>commons-logging</groupId>
<artifactId>commons-logging</artifactId>
</exclusion>
<exclusion>
<groupId>org.apache.commons</groupId>
<artifactId>commons-compress</artifactId>
</exclusion>
<exclusion>
<groupId>commons-codec</groupId>
<artifactId>commons-codec</artifactId>
</exclusion>
<exclusion>
<groupId>org.apache.avro</groupId>
<artifactId>avro</artifactId>
</exclusion>
<exclusion>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
</exclusion>
<exclusion>
<groupId>commons-lang</groupId>
<artifactId>commons-lang</artifactId>
</exclusion>
<exclusion>
<groupId>commons-cli</groupId>
<artifactId>commons-cli</artifactId>
</exclusion>
<exclusion>
<groupId>commons-collections</groupId>
<artifactId>commons-collections</artifactId>
</exclusion>
<exclusion>
<groupId>commons-io</groupId>
<artifactId>commons-io</artifactId>
</exclusion>
<exclusion>
<groupId>com.google.code.gson</groupId>
<artifactId>gson</artifactId>
</exclusion>
<exclusion>
<groupId>org.apache.httpcomponents</groupId>
<artifactId>httpclient</artifactId>
</exclusion>
<exclusion>
<groupId>org.apache.httpcomponents</groupId>
<artifactId>httpcore</artifactId>
</exclusion>
<exclusion>
<groupId>org.mortbay.jetty</groupId>
<artifactId>jetty</artifactId>
</exclusion>
<exclusion>
<groupId>org.mortbay.jetty</groupId>
<artifactId>jetty-util</artifactId>
</exclusion>
<exclusion>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
</exclusion>
<exclusion>
<groupId>com.google.protobuf</groupId>
<artifactId>protobuf-java</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-hadoop-libraries-nar</artifactId>
<version>2.0.0-SNAPSHOT</version>
<type>nar</type>
</dependency>
</dependencies>
</project>

319
nifi-nar-bundles/nifi-flume-bundle/nifi-flume-nar/src/main/resources/META-INF/LICENSE
View File

@ -1,319 +0,0 @@
Apache License
Version 2.0, January 2004
http://www.apache.org/licenses/
TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
1. Definitions.
"License" shall mean the terms and conditions for use, reproduction,
and distribution as defined by Sections 1 through 9 of this document.
"Licensor" shall mean the copyright owner or entity authorized by
the copyright owner that is granting the License.
"Legal Entity" shall mean the union of the acting entity and all
other entities that control, are controlled by, or are under common
control with that entity. For the purposes of this definition,
"control" means (i) the power, direct or indirect, to cause the
direction or management of such entity, whether by contract or
otherwise, or (ii) ownership of fifty percent (50%) or more of the
outstanding shares, or (iii) beneficial ownership of such entity.
"You" (or "Your") shall mean an individual or Legal Entity
exercising permissions granted by this License.
"Source" form shall mean the preferred form for making modifications,
including but not limited to software source code, documentation
source, and configuration files.
"Object" form shall mean any form resulting from mechanical
transformation or translation of a Source form, including but
not limited to compiled object code, generated documentation,
and conversions to other media types.
"Work" shall mean the work of authorship, whether in Source or
Object form, made available under the License, as indicated by a
copyright notice that is included in or attached to the work
(an example is provided in the Appendix below).
"Derivative Works" shall mean any work, whether in Source or Object
form, that is based on (or derived from) the Work and for which the
editorial revisions, annotations, elaborations, or other modifications
represent, as a whole, an original work of authorship. For the purposes
of this License, Derivative Works shall not include works that remain
separable from, or merely link (or bind by name) to the interfaces of,
the Work and Derivative Works thereof.
"Contribution" shall mean any work of authorship, including
the original version of the Work and any modifications or additions
to that Work or Derivative Works thereof, that is intentionally
submitted to Licensor for inclusion in the Work by the copyright owner
or by an individual or Legal Entity authorized to submit on behalf of
the copyright owner. For the purposes of this definition, "submitted"
means any form of electronic, verbal, or written communication sent
to the Licensor or its representatives, including but not limited to
communication on electronic mailing lists, source code control systems,
and issue tracking systems that are managed by, or on behalf of, the
Licensor for the purpose of discussing and improving the Work, but
excluding communication that is conspicuously marked or otherwise
designated in writing by the copyright owner as "Not a Contribution."
"Contributor" shall mean Licensor and any individual or Legal Entity
on behalf of whom a Contribution has been received by Licensor and
subsequently incorporated within the Work.
2. Grant of Copyright License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
copyright license to reproduce, prepare Derivative Works of,
publicly display, publicly perform, sublicense, and distribute the
Work and such Derivative Works in Source or Object form.
3. Grant of Patent License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
(except as stated in this section) patent license to make, have made,
use, offer to sell, sell, import, and otherwise transfer the Work,
where such license applies only to those patent claims licensable
by such Contributor that are necessarily infringed by their
Contribution(s) alone or by combination of their Contribution(s)
with the Work to which such Contribution(s) was submitted. If You
institute patent litigation against any entity (including a
cross-claim or counterclaim in a lawsuit) alleging that the Work
or a Contribution incorporated within the Work constitutes direct
or contributory patent infringement, then any patent licenses
granted to You under this License for that Work shall terminate
as of the date such litigation is filed.
4. Redistribution. You may reproduce and distribute copies of the
Work or Derivative Works thereof in any medium, with or without
modifications, and in Source or Object form, provided that You
meet the following conditions:
(a) You must give any other recipients of the Work or
Derivative Works a copy of this License; and
(b) You must cause any modified files to carry prominent notices
stating that You changed the files; and
(c) You must retain, in the Source form of any Derivative Works
that You distribute, all copyright, patent, trademark, and
attribution notices from the Source form of the Work,
excluding those notices that do not pertain to any part of
the Derivative Works; and
(d) If the Work includes a "NOTICE" text file as part of its
distribution, then any Derivative Works that You distribute must
include a readable copy of the attribution notices contained
within such NOTICE file, excluding those notices that do not
pertain to any part of the Derivative Works, in at least one
of the following places: within a NOTICE text file distributed
as part of the Derivative Works; within the Source form or
documentation, if provided along with the Derivative Works; or,
within a display generated by the Derivative Works, if and
wherever such third-party notices normally appear. The contents
of the NOTICE file are for informational purposes only and
do not modify the License. You may add Your own attribution
notices within Derivative Works that You distribute, alongside
or as an addendum to the NOTICE text from the Work, provided
that such additional attribution notices cannot be construed
as modifying the License.
You may add Your own copyright statement to Your modifications and
may provide additional or different license terms and conditions
for use, reproduction, or distribution of Your modifications, or
for any such Derivative Works as a whole, provided Your use,
reproduction, and distribution of the Work otherwise complies with
the conditions stated in this License.
5. Submission of Contributions. Unless You explicitly state otherwise,
any Contribution intentionally submitted for inclusion in the Work
by You to the Licensor shall be under the terms and conditions of
this License, without any additional terms or conditions.
Notwithstanding the above, nothing herein shall supersede or modify
the terms of any separate license agreement you may have executed
with Licensor regarding such Contributions.
6. Trademarks. This License does not grant permission to use the trade
names, trademarks, service marks, or product names of the Licensor,
except as required for reasonable and customary use in describing the
origin of the Work and reproducing the content of the NOTICE file.
7. Disclaimer of Warranty. Unless required by applicable law or
agreed to in writing, Licensor provides the Work (and each
Contributor provides its Contributions) on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
implied, including, without limitation, any warranties or conditions
of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
PARTICULAR PURPOSE. You are solely responsible for determining the
appropriateness of using or redistributing the Work and assume any
risks associated with Your exercise of permissions under this License.
8. Limitation of Liability. In no event and under no legal theory,
whether in tort (including negligence), contract, or otherwise,
unless required by applicable law (such as deliberate and grossly
negligent acts) or agreed to in writing, shall any Contributor be
liable to You for damages, including any direct, indirect, special,
incidental, or consequential damages of any character arising as a
result of this License or out of the use or inability to use the
Work (including but not limited to damages for loss of goodwill,
work stoppage, computer failure or malfunction, or any and all
other commercial damages or losses), even if such Contributor
has been advised of the possibility of such damages.
9. Accepting Warranty or Additional Liability. While redistributing
the Work or Derivative Works thereof, You may choose to offer,
and charge a fee for, acceptance of support, warranty, indemnity,
or other liability obligations and/or rights consistent with this
License. However, in accepting such obligations, You may act only
on Your own behalf and on Your sole responsibility, not on behalf
of any other Contributor, and only if You agree to indemnify,
defend, and hold each Contributor harmless for any liability
incurred by, or claims asserted against, such Contributor by reason
of your accepting any such warranty or additional liability.
END OF TERMS AND CONDITIONS
APPENDIX: How to apply the Apache License to your work.
To apply the Apache License to your work, attach the following
boilerplate notice, with the fields enclosed by brackets "[]"
replaced with your own identifying information. (Don't include
the brackets!) The text should be enclosed in the appropriate
comment syntax for the file format. We also recommend that a
file or class name and description of purpose be included on the
same "printed page" as the copyright notice for easier
identification within third-party archives.
Copyright [yyyy] [name of copyright owner]
Licensed 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.
APACHE NIFI SUBCOMPONENTS:
The Apache NiFi project contains subcomponents with separate copyright
notices and license terms. Your use of the source code for the these
subcomponents is subject to the terms and conditions of the following
licenses.
The binary distribution of this product bundles 'SUAsync Library' which is
available under a 3-Clause BSD License.
Copyright (c) 2010 StumbleUpon, Inc. All rights reserved.
Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions are met:
- Redistributions of source code must retain the above copyright notice,
this list of conditions and the following disclaimer.
- Redistributions in binary form must reproduce the above copyright notice,
this list of conditions and the following disclaimer in the documentation
and/or other materials provided with the distribution.
- Neither the name of the StumbleUpon nor the names of its contributors
may be used to endorse or promote products derived from this software
without specific prior written permission.
THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
POSSIBILITY OF SUCH DAMAGE.
The binary distribution of this product bundles 'Asynchronous HBase Client'
which is available under a 3-Clause BSD License.
Copyright (C) 2010-2012 The Async HBase Authors. All rights reserved.
Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions are met:
- Redistributions of source code must retain the above copyright notice,
this list of conditions and the following disclaimer.
- Redistributions in binary form must reproduce the above copyright notice,
this list of conditions and the following disclaimer in the documentation
and/or other materials provided with the distribution.
- Neither the name of the StumbleUpon nor the names of its contributors
may be used to endorse or promote products derived from this software
without specific prior written permission.
THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
POSSIBILITY OF SUCH DAMAGE.
The binary distribution of this product bundles 'JOpt Simple' which is
available under the MIT license.
The MIT License
Copyright (c) 2004-2011 Paul R. Holser, Jr.
Permission is hereby granted, free of charge, to any person obtaining
a copy of this software and associated documentation files (the
"Software"), to deal in the Software without restriction, including
without limitation the rights to use, copy, modify, merge, publish,
distribute, sublicense, and/or sell copies of the Software, and to
permit persons to whom the Software is furnished to do so, subject to
the following conditions:
The above copyright notice and this permission notice shall be
included in all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
The binary distribution of this product bundles 'Scala Library' under a BSD
style license.
Copyright (c) 2002-2015 EPFL
Copyright (c) 2011-2015 Typesafe, Inc.
All rights reserved.
Redistribution and use in source and binary forms, with or without modification,
are permitted provided that the following conditions are met:
Redistributions of source code must retain the above copyright notice, this list of
conditions and the following disclaimer.
Redistributions in binary form must reproduce the above copyright notice, this list of
conditions and the following disclaimer in the documentation and/or other materials
provided with the distribution.
Neither the name of the EPFL nor the names of its contributors may be used to endorse
or promote products derived from this software without specific prior written permission.
THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS “AS IS” AND ANY EXPRESS
OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY
AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR
CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER
IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT
OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

186
nifi-nar-bundles/nifi-flume-bundle/nifi-flume-nar/src/main/resources/META-INF/NOTICE
View File

@ -1,186 +0,0 @@
nifi-flume-nar
Copyright 2015-2020 The Apache Software Foundation
This product includes software developed at
The Apache Software Foundation (http://www.apache.org/).
******************
Apache Software License v2
******************
The following binary components are provided under the Apache Software License v2
(ASLv2) Apache Avro
The following NOTICE information applies:
Apache Avro
Copyright 2009-2017 The Apache Software Foundation
(ASLv2) Apache Commons JEXL
The following NOTICE information applies:
Apache Commons JEXL
Copyright 2001-2011 The Apache Software Foundation
(ASLv2) Apache Commons Lang
The following NOTICE information applies:
Apache Commons Lang
Copyright 2001-2015 The Apache Software Foundation
This product includes software from the Spring Framework,
under the Apache License 2.0 (see: StringUtils.containsWhitespace())
(ASLv2) Apache Flume
The following NOTICE information applies:
Apache Flume
Copyright 2011-2015 Apache Software Foundation
asynchbase is BSD-licensed software (https://github.com/OpenTSDB/asynchbase)
async is BSD-licensed software (https://github.com/stumbleupon/async)
jopt-simple is MIT licensed software (http://pholser.github.io/jopt-simple/license.html)
scala-library is BSD-like licensed software (http://www.scala-lang.org/license.html)
(ASLv2) Xalan
This product includes software developed by
The Apache Software Foundation (http://www.apache.org/).
Portions of this software was originally based on the following:
- software copyright (c) 1999-2002, Lotus Development Corporation., http://www.lotus.com.
- software copyright (c) 2001-2002, Sun Microsystems., http://www.sun.com.
- software copyright (c) 2003, IBM Corporation., http://www.ibm.com.
- voluntary contributions made by Ovidiu Predescu (ovidiu@cup.hp.com) on behalf of the
Apache Software Foundation and was originally developed at Hewlett Packard Company.
(ASLv2) Apache Xerces Java
Copyright 1999-2007 The Apache Software Foundation
This product includes software developed at
The Apache Software Foundation (http://www.apache.org/).
Portions of this software were originally based on the following:
- software copyright (c) 1999, IBM Corporation., http://www.ibm.com.
- software copyright (c) 1999, Sun Microsystems., http://www.sun.com.
- voluntary contributions made by Paul Eng on behalf of the
Apache Software Foundation that were originally developed at iClick, Inc.,
software copyright (c) 1999.
(ASLv2) Apache XML Commons XML APIs
Copyright 2006 The Apache Software Foundation.
This product includes software developed at
The Apache Software Foundation (http://www.apache.org/).
Portions of this software were originally based on the following:
- software copyright (c) 1999, IBM Corporation., http://www.ibm.com.
- software copyright (c) 1999, Sun Microsystems., http://www.sun.com.
- software copyright (c) 2000 World Wide Web Consortium, http://www.w3.org
(ASLv2) IRClib
The following NOTICE information applies:
IRClib -- A Java Internet Relay Chat library --
Copyright (C) 2002 - 2006 Christoph Schwering <schwering@gmail.com>
(ASLv2) Jackson JSON processor
The following NOTICE information applies:
# Jackson JSON processor
Jackson is a high-performance, Free/Open Source JSON processing library.
It was originally written by Tatu Saloranta (tatu.saloranta@iki.fi), and has
been in development since 2007.
It is currently developed by a community of developers, as well as supported
commercially by FasterXML.com.
## Licensing
Jackson core and extension components may licensed under different licenses.
To find the details that apply to this artifact see the accompanying LICENSE file.
For more information, including possible other licensing options, contact
FasterXML.com (http://fasterxml.com).
## Credits
A list of contributors may be found from CREDITS file, which is included
in some artifacts (usually source distributions); but is always available
from the source code management (SCM) system project uses.
(ASLv2) Joda-Time
The following NOTICE information applies:
=============================================================================
= NOTICE file corresponding to section 4d of the Apache License Version 2.0 =
=============================================================================
This product includes software developed by
Joda.org (http://www.joda.org/).
(ASLv2) Apache Kafka
The following NOTICE information applies:
Apache Kafka
Copyright 2012 The Apache Software Foundation.
(ASLv2) Kite SDK
The following NOTICE information applies:
This product includes software developed by Cloudera, Inc.
(http://www.cloudera.com/).
This product includes software developed at
The Apache Software Foundation (http://www.apache.org/).
This product includes software developed by
Saxonica (http://www.saxonica.com/).
(ASLv2) Apache Thrift
The following NOTICE information applies:
Apache Thrift
Copyright 2006-2010 The Apache Software Foundation.
(ASLv2) Yammer Metrics
The following NOTICE information applies:
Metrics
Copyright 2010-2012 Coda Hale and Yammer, Inc.
This product includes software developed by Coda Hale and Yammer, Inc.
This product includes code derived from the JSR-166 project (ThreadLocalRandom), which was released
with the following comments:
Written by Doug Lea with assistance from members of JCP JSR-166
Expert Group and released to the public domain, as explained at
http://creativecommons.org/publicdomain/zero/1.0/
(ASLv2) Apache MINA
The following NOTICE information applies:
Apache MINA Core
Copyright 2004-2011 Apache MINA Project
(ASLv2) The Netty Project
The following NOTICE information applies:
The Netty Project
Copyright 2011 The Netty Project
(ASLv2) opencsv (net.sf.opencsv:opencsv:2.3)
(ASLv2) Parquet MR
The following NOTICE information applies:
Parquet MR
Copyright 2012 Twitter, Inc.
This project includes code from https://github.com/lemire/JavaFastPFOR
parquet-column/src/main/java/parquet/column/values/bitpacking/LemireBitPacking.java
Apache License Version 2.0 http://www.apache.org/licenses/.
(c) Daniel Lemire, http://lemire.me/en/
(ASLv2) Jetty
The following NOTICE information applies:
Jetty Web Container
Copyright 1995-2019 Mort Bay Consulting Pty Ltd.
(ASLv2) Apache Velocity
The following NOTICE information applies:
Apache Velocity
Copyright (C) 2000-2007 The Apache Software Foundation
(ASLv2) ZkClient
The following NOTICE information applies:
ZkClient
Copyright 2009 Stefan Groschupf

219
nifi-nar-bundles/nifi-flume-bundle/nifi-flume-processors/pom.xml
View File

@ -1,219 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
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.
-->
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-flume-bundle</artifactId>
<version>2.0.0-SNAPSHOT</version>
</parent>
<artifactId>nifi-flume-processors</artifactId>
<packaging>jar</packaging>
<properties>
<flume.version>1.11.0</flume.version>
</properties>
<dependencies>
<dependency>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-api</artifactId>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>log4j-over-slf4j</artifactId>
</dependency>
<dependency>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-utils</artifactId>
<version>2.0.0-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-flowfile-packager</artifactId>
<version>2.0.0-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>org.apache.flume</groupId>
<artifactId>flume-ng-sdk</artifactId>
<version>${flume.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flume</groupId>
<artifactId>flume-ng-core</artifactId>
<version>${flume.version}</version>
<exclusions>
<exclusion>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
</exclusion>
<exclusion>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
</exclusion>
<exclusion>
<groupId>commons-logging</groupId>
<artifactId>commons-logging</artifactId>
</exclusion>
<!-- not allowed findbugs version pulled in by guava version - this should be fixed by flume... -->
<exclusion>
<groupId>com.google.code.findbugs</groupId>
<artifactId>jsr305</artifactId>
</exclusion>
</exclusions>
</dependency>
<!-- pulled in to replace the not allowed version of findbugs -->
<dependency>
<groupId>com.github.stephenc.findbugs</groupId>
<artifactId>findbugs-annotations</artifactId>
<version>1.3.9-1</version>
</dependency>
<!-- Flume Sources -->
<dependency>
<groupId>org.apache.flume.flume-ng-sources</groupId>
<artifactId>flume-jms-source</artifactId>
<version>${flume.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flume.flume-ng-sources</groupId>
<artifactId>flume-kafka-source</artifactId>
<version>${flume.version}</version>
<exclusions>
<exclusion>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.apache.flume.flume-ng-sources</groupId>
<artifactId>flume-scribe-source</artifactId>
<version>${flume.version}</version>
</dependency>
<!-- Flume Sinks -->
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
</dependency>
<dependency>
<groupId>org.apache.flume.flume-ng-sinks</groupId>
<artifactId>flume-hdfs-sink</artifactId>
<version>${flume.version}</version>
</dependency>
<!-- HDFS sink dependencies -->
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-common</artifactId>
<version>${hadoop.version}</version>
<scope>provided</scope>
<exclusions>
<exclusion>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-reload4j</artifactId>
</exclusion>
<exclusion>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
</exclusion>
<exclusion>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
</exclusion>
<exclusion>
<groupId>commons-logging</groupId>
<artifactId>commons-logging</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-hdfs</artifactId>
<version>${hadoop.version}</version>
<scope>provided</scope>
<exclusions>
<exclusion>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
</exclusion>
<exclusion>
<groupId>commons-logging</groupId>
<artifactId>commons-logging</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.apache.flume.flume-ng-sinks</groupId>
<artifactId>flume-hive-sink</artifactId>
<version>${flume.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flume.flume-ng-sinks</groupId>
<artifactId>flume-irc-sink</artifactId>
<version>${flume.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flume.flume-ng-sinks</groupId>
<artifactId>flume-ng-hbase-sink</artifactId>
<version>${flume.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flume.flume-ng-sinks</groupId>
<artifactId>flume-ng-kafka-sink</artifactId>
<version>${flume.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flume.flume-ng-sinks</groupId>
<artifactId>flume-ng-morphline-solr-sink</artifactId>
<version>${flume.version}</version>
</dependency>
<dependency>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-mock</artifactId>
<version>2.0.0-SNAPSHOT</version>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.rat</groupId>
<artifactId>apache-rat-plugin</artifactId>
<configuration>
<excludes>
<!-- test data -->
<exclude>src/test/resources/testdata/*</exclude>
</excludes>
</configuration>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-surefire-plugin</artifactId>
<configuration>
<systemPropertyVariables>
<java.io.tmpdir>${project.build.directory}</java.io.tmpdir>
</systemPropertyVariables>
</configuration>
</plugin>
</plugins>
</build>
</project>

160
nifi-nar-bundles/nifi-flume-bundle/nifi-flume-processors/src/main/java/org/apache/nifi/processors/flume/AbstractFlumeProcessor.java
View File

@ -1,160 +0,0 @@
/*
* 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.flume;
import com.google.common.collect.Maps;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.SinkFactory;
import org.apache.flume.SourceFactory;
import org.apache.flume.sink.DefaultSinkFactory;
import org.apache.flume.source.DefaultSourceFactory;
import org.apache.nifi.components.ValidationContext;
import org.apache.nifi.components.ValidationResult;
import org.apache.nifi.components.Validator;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.processor.AbstractSessionFactoryProcessor;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.ProcessSessionFactory;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.io.OutputStreamCallback;
import org.apache.nifi.processors.flume.util.FlowFileEvent;
import java.io.IOException;
import java.io.OutputStream;
import java.io.StringReader;
import java.util.Map;
import java.util.Properties;
/**
* This is a base class that is helpful when building processors interacting with Flume.
*/
public abstract class AbstractFlumeProcessor extends AbstractSessionFactoryProcessor {
protected static final SourceFactory SOURCE_FACTORY = new DefaultSourceFactory();
protected static final SinkFactory SINK_FACTORY = new DefaultSinkFactory();
protected static Event flowFileToEvent(FlowFile flowFile, ProcessSession session) {
return new FlowFileEvent(flowFile, session);
}
protected static void transferEvent(final Event event, ProcessSession session,
Relationship relationship) {
FlowFile flowFile = session.create();
flowFile = session.putAllAttributes(flowFile, event.getHeaders());
flowFile = session.write(flowFile, new OutputStreamCallback() {
@Override
public void process(final OutputStream out) throws IOException {
out.write(event.getBody());
}
});
session.getProvenanceReporter()
.create(flowFile);
session.transfer(flowFile, relationship);
}
protected static Validator createSourceValidator() {
return new Validator() {
@Override
public ValidationResult validate(final String subject, final String value, final ValidationContext context) {
String reason = null;
try {
ExecuteFlumeSource.SOURCE_FACTORY.create("NiFi Source", value);
} catch (Exception ex) {
reason = ex.getLocalizedMessage();
reason = Character.toLowerCase(reason.charAt(0)) + reason.substring(1);
}
return new ValidationResult.Builder().subject(subject)
.input(value)
.explanation(reason)
.valid(reason == null)
.build();
}
};
}
protected static Validator createSinkValidator() {
return new Validator() {
@Override
public ValidationResult validate(final String subject, final String value, final ValidationContext context) {
String reason = null;
try {
ExecuteFlumeSink.SINK_FACTORY.create("NiFi Sink", value);
} catch (Exception ex) {
reason = ex.getLocalizedMessage();
reason = Character.toLowerCase(reason.charAt(0)) + reason.substring(1);
}
return new ValidationResult.Builder().subject(subject)
.input(value)
.explanation(reason)
.valid(reason == null)
.build();
}
};
}
protected static Context getFlumeContext(String flumeConfig, String prefix) {
Properties flumeProperties = new Properties();
if (flumeConfig != null) {
try {
flumeProperties.load(new StringReader(flumeConfig));
} catch (IOException ex) {
throw new RuntimeException(ex);
}
}
Map<String, String> parameters = Maps.newHashMap();
for (String property : flumeProperties.stringPropertyNames()) {
parameters.put(property, flumeProperties.getProperty(property));
}
return new Context(new Context(parameters).getSubProperties(prefix));
}
protected static Context getFlumeSourceContext(String flumeConfig,
String agentName, String sourceName) {
return getFlumeContext(flumeConfig, agentName + ".sources." + sourceName + ".");
}
protected static Context getFlumeSinkContext(String flumeConfig,
String agentName, String sinkName) {
return getFlumeContext(flumeConfig, agentName + ".sinks." + sinkName + ".");
}
/*
* Borrowed from AbstractProcessor. The FlumeSourceProcessor needs to implement this directly
* to handle event driven sources, but it's marked final in AbstractProcessor.
*/
@Override
public void onTrigger(final ProcessContext context, final ProcessSessionFactory sessionFactory) throws ProcessException {
final ProcessSession session = sessionFactory.createSession();
try {
onTrigger(context, session);
session.commitAsync();
} catch (final Throwable t) {
getLogger()
.error("{} failed to process due to {}; rolling back session", new Object[]{this, t});
session.rollback(true);
throw t;
}
}
public abstract void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException;
}

158
nifi-nar-bundles/nifi-flume-bundle/nifi-flume-processors/src/main/java/org/apache/nifi/processors/flume/ExecuteFlumeSink.java
View File

@ -1,158 +0,0 @@
/*
* 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.flume;
import com.google.common.base.Throwables;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.Sink;
import org.apache.flume.conf.Configurables;
import org.apache.nifi.annotation.behavior.InputRequirement;
import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
import org.apache.nifi.annotation.behavior.Restricted;
import org.apache.nifi.annotation.behavior.Restriction;
import org.apache.nifi.annotation.behavior.TriggerSerially;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.annotation.documentation.DeprecationNotice;
import org.apache.nifi.annotation.documentation.Tags;
import org.apache.nifi.annotation.lifecycle.OnScheduled;
import org.apache.nifi.annotation.lifecycle.OnStopped;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.components.RequiredPermission;
import org.apache.nifi.components.Validator;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.ProcessorInitializationContext;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.util.StandardValidators;
import java.util.List;
import java.util.Set;
/**
* This processor runs a Flume sink
*/
@TriggerSerially
@Tags({"flume", "hadoop", "put", "sink"})
@InputRequirement(Requirement.INPUT_REQUIRED)
@CapabilityDescription("Execute a Flume sink. Each input FlowFile is converted into a Flume Event for processing by the sink.")
@Restricted(
restrictions = {
@Restriction(
requiredPermission = RequiredPermission.EXECUTE_CODE,
explanation = "Provides operator the ability to execute arbitrary Flume configurations assuming all permissions that NiFi has.")
}
)
@DeprecationNotice(reason = "Apache Flume pipelines should be implemented using Apache NiFi components")
public class ExecuteFlumeSink extends AbstractFlumeProcessor {
public static final PropertyDescriptor SINK_TYPE = new PropertyDescriptor.Builder()
.name("Sink Type")
.description("The component type name for the sink. For some sinks, this is a short, symbolic name (e.g. hdfs)."
+ " For others, it's the fully-qualified name of the Sink class. See the Flume User Guide for details.")
.required(true)
.addValidator(createSinkValidator())
.build();
public static final PropertyDescriptor AGENT_NAME = new PropertyDescriptor.Builder()
.name("Agent Name")
.description("The name of the agent used in the Flume sink configuration")
.required(true)
.defaultValue("tier1")
.addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
.build();
public static final PropertyDescriptor SOURCE_NAME = new PropertyDescriptor.Builder()
.name("Sink Name")
.description("The name of the sink used in the Flume sink configuration")
.required(true)
.defaultValue("sink-1")
.addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
.build();
public static final PropertyDescriptor FLUME_CONFIG = new PropertyDescriptor.Builder()
.name("Flume Configuration")
.description("The Flume configuration for the sink copied from the flume.properties file")
.required(true)
.defaultValue("")
.addValidator(Validator.VALID)
.build();
public static final Relationship SUCCESS = new Relationship.Builder().name("success").build();
public static final Relationship FAILURE = new Relationship.Builder().name("failure").build();
private List<PropertyDescriptor> descriptors;
private Set<Relationship> relationships;
private volatile Sink sink;
private volatile NifiSinkSessionChannel channel;
@Override
protected void init(final ProcessorInitializationContext context) {
this.descriptors = ImmutableList.of(SINK_TYPE, AGENT_NAME, SOURCE_NAME, FLUME_CONFIG);
this.relationships = ImmutableSet.of(SUCCESS, FAILURE);
}
@Override
protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
return descriptors;
}
@Override
public Set<Relationship> getRelationships() {
return relationships;
}
@OnScheduled
public void onScheduled(final ProcessContext context) {
try {
channel = new NifiSinkSessionChannel(SUCCESS, FAILURE);
channel.start();
sink = SINK_FACTORY.create(context.getProperty(SOURCE_NAME).getValue(),
context.getProperty(SINK_TYPE).getValue());
sink.setChannel(channel);
String flumeConfig = context.getProperty(FLUME_CONFIG).getValue();
String agentName = context.getProperty(AGENT_NAME).getValue();
String sinkName = context.getProperty(SOURCE_NAME).getValue();
Configurables.configure(sink,
getFlumeSinkContext(flumeConfig, agentName, sinkName));
sink.start();
} catch (Throwable th) {
getLogger().error("Error creating sink", th);
throw Throwables.propagate(th);
}
}
@OnStopped
public void stopped() {
sink.stop();
channel.stop();
}
@Override
public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
channel.setSession(session);
try {
sink.process();
} catch (EventDeliveryException ex) {
throw new ProcessException("Flume event delivery failed", ex);
}
}
}

210
nifi-nar-bundles/nifi-flume-bundle/nifi-flume-processors/src/main/java/org/apache/nifi/processors/flume/ExecuteFlumeSource.java
View File

@ -1,210 +0,0 @@
/*
* 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.flume;
import com.google.common.base.Throwables;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.EventDrivenSource;
import org.apache.flume.PollableSource;
import org.apache.flume.Source;
import org.apache.flume.channel.ChannelProcessor;
import org.apache.flume.conf.Configurables;
import org.apache.flume.source.EventDrivenSourceRunner;
import org.apache.nifi.annotation.behavior.InputRequirement;
import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
import org.apache.nifi.annotation.behavior.Restricted;
import org.apache.nifi.annotation.behavior.Restriction;
import org.apache.nifi.annotation.behavior.TriggerSerially;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.annotation.documentation.DeprecationNotice;
import org.apache.nifi.annotation.documentation.Tags;
import org.apache.nifi.annotation.lifecycle.OnScheduled;
import org.apache.nifi.annotation.lifecycle.OnStopped;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.components.RequiredPermission;
import org.apache.nifi.components.Validator;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.ProcessSessionFactory;
import org.apache.nifi.processor.ProcessorInitializationContext;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.util.StandardValidators;
import java.util.List;
import java.util.Set;
import java.util.concurrent.atomic.AtomicReference;
/**
* This processor runs a Flume source
*/
@TriggerSerially
@Tags({"flume", "hadoop", "get", "source"})
@InputRequirement(Requirement.INPUT_FORBIDDEN)
@CapabilityDescription("Execute a Flume source. Each Flume Event is sent to the success relationship as a FlowFile")
@Restricted(
restrictions = {
@Restriction(
requiredPermission = RequiredPermission.EXECUTE_CODE,
explanation = "Provides operator the ability to execute arbitrary Flume configurations assuming all permissions that NiFi has.")
}
)
@DeprecationNotice(reason = "Apache Flume pipelines should be implemented using Apache NiFi components")
public class ExecuteFlumeSource extends AbstractFlumeProcessor {
public static final PropertyDescriptor SOURCE_TYPE = new PropertyDescriptor.Builder()
.name("Source Type")
.description("The component type name for the source. For some sources, this is a short, symbolic name"
+ " (e.g. spooldir). For others, it's the fully-qualified name of the Source class. See the Flume User Guide for details.")
.required(true)
.addValidator(createSourceValidator())
.build();
public static final PropertyDescriptor AGENT_NAME = new PropertyDescriptor.Builder()
.name("Agent Name")
.description("The name of the agent used in the Flume source configuration")
.required(true)
.defaultValue("tier1")
.addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
.build();
public static final PropertyDescriptor SOURCE_NAME = new PropertyDescriptor.Builder()
.name("Source Name")
.description("The name of the source used in the Flume source configuration")
.required(true)
.defaultValue("src-1")
.addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
.build();
public static final PropertyDescriptor FLUME_CONFIG = new PropertyDescriptor.Builder()
.name("Flume Configuration")
.description("The Flume configuration for the source copied from the flume.properties file")
.required(true)
.defaultValue("")
.addValidator(Validator.VALID)
.build();
public static final Relationship SUCCESS = new Relationship.Builder().name("success")
.build();
private List<PropertyDescriptor> descriptors;
private Set<Relationship> relationships;
private volatile Source source;
private final NifiSessionChannel pollableSourceChannel = new NifiSessionChannel(SUCCESS);
private final AtomicReference<ProcessSessionFactory> sessionFactoryRef = new AtomicReference<>(null);
private final AtomicReference<EventDrivenSourceRunner> runnerRef = new AtomicReference<>(null);
private final AtomicReference<NifiSessionFactoryChannel> eventDrivenSourceChannelRef = new AtomicReference<>(null);
@Override
protected void init(final ProcessorInitializationContext context) {
this.descriptors = ImmutableList.of(SOURCE_TYPE, AGENT_NAME, SOURCE_NAME, FLUME_CONFIG);
this.relationships = ImmutableSet.of(SUCCESS);
}
@Override
protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
return descriptors;
}
@Override
public Set<Relationship> getRelationships() {
return relationships;
}
@OnScheduled
public void onScheduled(final ProcessContext context) {
try {
source = SOURCE_FACTORY.create(
context.getProperty(SOURCE_NAME).getValue(),
context.getProperty(SOURCE_TYPE).getValue());
String flumeConfig = context.getProperty(FLUME_CONFIG).getValue();
String agentName = context.getProperty(AGENT_NAME).getValue();
String sourceName = context.getProperty(SOURCE_NAME).getValue();
Configurables.configure(source,
getFlumeSourceContext(flumeConfig, agentName, sourceName));
if (source instanceof PollableSource) {
source.setChannelProcessor(new ChannelProcessor(
new NifiChannelSelector(pollableSourceChannel)));
source.start();
}
} catch (Throwable th) {
getLogger().error("Error creating source", th);
throw Throwables.propagate(th);
}
}
@OnStopped
public void stopped() {
if (source instanceof PollableSource) {
source.stop();
} else {
EventDrivenSourceRunner runner = runnerRef.get();
if (runner != null) {
runner.stop();
runnerRef.compareAndSet(runner, null);
}
NifiSessionFactoryChannel eventDrivenSourceChannel = eventDrivenSourceChannelRef.get();
if (eventDrivenSourceChannel != null) {
eventDrivenSourceChannel.stop();
eventDrivenSourceChannelRef.compareAndSet(eventDrivenSourceChannel, null);
}
}
sessionFactoryRef.set(null);
}
@Override
public void onTrigger(ProcessContext context, ProcessSessionFactory sessionFactory) throws ProcessException {
if (source instanceof PollableSource) {
super.onTrigger(context, sessionFactory);
} else if (source instanceof EventDrivenSource) {
ProcessSessionFactory old = sessionFactoryRef.getAndSet(sessionFactory);
if (old != sessionFactory) {
if (runnerRef.get() != null) {
stopped();
sessionFactoryRef.set(sessionFactory);
}
runnerRef.set(new EventDrivenSourceRunner());
eventDrivenSourceChannelRef.set(new NifiSessionFactoryChannel(sessionFactoryRef.get(), SUCCESS));
eventDrivenSourceChannelRef.get().start();
source.setChannelProcessor(new ChannelProcessor(
new NifiChannelSelector(eventDrivenSourceChannelRef.get())));
runnerRef.get().setSource(source);
runnerRef.get().start();
}
}
}
@Override
public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
if (source instanceof PollableSource) {
PollableSource pollableSource = (PollableSource) source;
try {
pollableSourceChannel.setSession(session);
pollableSource.process();
} catch (EventDeliveryException ex) {
throw new ProcessException("Error processing pollable source", ex);
}
} else {
throw new ProcessException("Invalid source type: " + source.getClass().getSimpleName());
}
}
}

69
nifi-nar-bundles/nifi-flume-bundle/nifi-flume-processors/src/main/java/org/apache/nifi/processors/flume/NifiChannelSelector.java
View File

@ -1,69 +0,0 @@
/*
* 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.flume;
import com.google.common.collect.ImmutableList;
import java.util.List;
import org.apache.flume.Channel;
import org.apache.flume.ChannelSelector;
import org.apache.flume.Context;
import org.apache.flume.Event;
public class NifiChannelSelector implements ChannelSelector {
private String name;
private final List<Channel> requiredChannels;
private final List<Channel> optionalChannels;
public NifiChannelSelector(Channel channel) {
requiredChannels = ImmutableList.of(channel);
optionalChannels = ImmutableList.of();
}
@Override
public List<Channel> getRequiredChannels(Event event) {
return requiredChannels;
}
@Override
public List<Channel> getOptionalChannels(Event event) {
return optionalChannels;
}
@Override
public List<Channel> getAllChannels() {
return requiredChannels;
}
@Override
public void setChannels(List<Channel> channels) {
}
@Override
public String getName() {
return name;
}
@Override
public void setName(String name) {
this.name = name;
}
@Override
public void configure(Context context) {
}
}

47
nifi-nar-bundles/nifi-flume-bundle/nifi-flume-processors/src/main/java/org/apache/nifi/processors/flume/NifiSessionChannel.java
View File

@ -1,47 +0,0 @@
/*
* 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.flume;
import org.apache.flume.Context;
import org.apache.flume.channel.BasicChannelSemantics;
import org.apache.flume.channel.BasicTransactionSemantics;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.Relationship;
public class NifiSessionChannel extends BasicChannelSemantics {
private ProcessSession session;
private final Relationship relationship;
public NifiSessionChannel(Relationship relationship) {
this.relationship = relationship;
}
public void setSession(ProcessSession session) {
this.session = session;
}
@Override
protected BasicTransactionSemantics createTransaction() {
return new NifiTransaction(session, relationship);
}
@Override
public void configure(Context context) {
}
}

50
nifi-nar-bundles/nifi-flume-bundle/nifi-flume-processors/src/main/java/org/apache/nifi/processors/flume/NifiSessionFactoryChannel.java
View File

@ -1,50 +0,0 @@
/*
* 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.flume;
import org.apache.flume.ChannelFullException;
import org.apache.flume.Context;
import org.apache.flume.channel.BasicChannelSemantics;
import org.apache.flume.channel.BasicTransactionSemantics;
import org.apache.flume.lifecycle.LifecycleState;
import org.apache.nifi.processor.ProcessSessionFactory;
import org.apache.nifi.processor.Relationship;
public class NifiSessionFactoryChannel extends BasicChannelSemantics {
private final ProcessSessionFactory sessionFactory;
private final Relationship relationship;
public NifiSessionFactoryChannel(ProcessSessionFactory sessionFactory, Relationship relationship) {
this.sessionFactory = sessionFactory;
this.relationship = relationship;
}
@Override
protected BasicTransactionSemantics createTransaction() {
LifecycleState lifecycleState = getLifecycleState();
if (lifecycleState == LifecycleState.STOP) {
throw new ChannelFullException("Can't write to a stopped channel");
}
return new NifiTransaction(sessionFactory.createSession(), relationship);
}
@Override
public void configure(Context context) {
}
}

49
nifi-nar-bundles/nifi-flume-bundle/nifi-flume-processors/src/main/java/org/apache/nifi/processors/flume/NifiSinkSessionChannel.java
View File

@ -1,49 +0,0 @@
/*
* 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.flume;
import org.apache.flume.Context;
import org.apache.flume.channel.BasicChannelSemantics;
import org.apache.flume.channel.BasicTransactionSemantics;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.Relationship;
public class NifiSinkSessionChannel extends BasicChannelSemantics {
private ProcessSession session;
private final Relationship success;
private final Relationship failure;
public NifiSinkSessionChannel(Relationship success, Relationship failure) {
this.success = success;
this.failure = failure;
}
public void setSession(ProcessSession session) {
this.session = session;
}
@Override
protected BasicTransactionSemantics createTransaction() {
return new NifiSinkTransaction(session, success, failure);
}
@Override
public void configure(Context context) {
}
}

72
nifi-nar-bundles/nifi-flume-bundle/nifi-flume-processors/src/main/java/org/apache/nifi/processors/flume/NifiSinkTransaction.java
View File

@ -1,72 +0,0 @@
/*
* 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.flume;
import org.apache.flume.Event;
import org.apache.flume.channel.BasicTransactionSemantics;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processors.flume.util.FlowFileEvent;
import java.util.ArrayList;
import java.util.List;
class NifiSinkTransaction extends BasicTransactionSemantics {
private final ProcessSession session;
private final Relationship success;
private final Relationship failure;
private final List<FlowFile> flowFiles;
public NifiSinkTransaction(ProcessSession session, Relationship success, Relationship failure) {
this.session = session;
this.success = success;
this.failure = failure;
this.flowFiles = new ArrayList<>();
}
@Override
protected void doPut(Event event) throws InterruptedException {
AbstractFlumeProcessor.transferEvent(event, session, success);
}
@Override
protected Event doTake() throws InterruptedException {
FlowFile flowFile = session.get();
if (flowFile == null) {
return null;
}
flowFiles.add(flowFile);
return new FlowFileEvent(flowFile, session);
}
@Override
protected void doCommit() throws InterruptedException {
session.transfer(flowFiles, success);
session.commitAsync();
}
@Override
protected void doRollback() throws InterruptedException {
session.transfer(flowFiles, failure);
session.commitAsync();
}
}

54
nifi-nar-bundles/nifi-flume-bundle/nifi-flume-processors/src/main/java/org/apache/nifi/processors/flume/NifiTransaction.java
View File

@ -1,54 +0,0 @@
/*
* 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.flume;
import org.apache.flume.Event;
import org.apache.flume.channel.BasicTransactionSemantics;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.Relationship;
class NifiTransaction extends BasicTransactionSemantics {
private final ProcessSession session;
private final Relationship relationship;
public NifiTransaction(ProcessSession session, Relationship relationship) {
this.session = session;
this.relationship = relationship;
}
@Override
protected void doPut(Event event) throws InterruptedException {
AbstractFlumeProcessor.transferEvent(event, session, relationship);
}
@Override
protected Event doTake() throws InterruptedException {
throw new UnsupportedOperationException("Only put supported");
}
@Override
protected void doCommit() throws InterruptedException {
session.commitAsync();
}
@Override
protected void doRollback() throws InterruptedException {
session.rollback();
}
}

119
nifi-nar-bundles/nifi-flume-bundle/nifi-flume-processors/src/main/java/org/apache/nifi/processors/flume/util/FlowFileEvent.java
View File

@ -1,119 +0,0 @@
/*
* 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.flume.util;
import static org.apache.nifi.processors.flume.util.FlowFileEventConstants.ENTRY_DATE_HEADER;
import static org.apache.nifi.processors.flume.util.FlowFileEventConstants.ID_HEADER;
import static org.apache.nifi.processors.flume.util.FlowFileEventConstants.LAST_QUEUE_DATE_HEADER;
import static org.apache.nifi.processors.flume.util.FlowFileEventConstants.LINEAGE_START_DATE_HEADER;
import static org.apache.nifi.processors.flume.util.FlowFileEventConstants.SIZE_HEADER;
import com.google.common.collect.Maps;
import java.io.BufferedInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.Arrays;
import java.util.Map;
import org.apache.flume.Event;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.io.InputStreamCallback;
import org.apache.nifi.stream.io.StreamUtils;
public class FlowFileEvent implements Event {
private final FlowFile flowFile;
private final ProcessSession session;
private final Map<String, String> headers;
private boolean headersLoaded;
private final Object bodyLock;
private byte[] body;
private boolean bodyLoaded;
public FlowFileEvent(FlowFile flowFile, ProcessSession session) {
this.flowFile = flowFile;
this.session = session;
headers = Maps.newHashMap();
bodyLock = new Object();
bodyLoaded = false;
}
@Override
public Map<String, String> getHeaders() {
synchronized (headers) {
if (!headersLoaded) {
headers.putAll(flowFile.getAttributes());
headers.put(ENTRY_DATE_HEADER, Long.toString(flowFile.getEntryDate()));
headers.put(ID_HEADER, Long.toString(flowFile.getId()));
headers.put(LAST_QUEUE_DATE_HEADER, Long.toString(flowFile.getLastQueueDate()));
headers.put(LINEAGE_START_DATE_HEADER, Long.toString(flowFile.getLineageStartDate()));
headers.put(SIZE_HEADER, Long.toString(flowFile.getSize()));
headersLoaded = true;
}
}
return headers;
}
@Override
public void setHeaders(Map<String, String> headers) {
synchronized (this.headers) {
this.headers.clear();
this.headers.putAll(headers);
headersLoaded = true;
}
}
@Override
public byte[] getBody() {
synchronized (bodyLock) {
if (!bodyLoaded) {
if (flowFile.getSize() > Integer.MAX_VALUE) {
throw new RuntimeException("Can't get body of Event because the backing FlowFile is too large (" + flowFile.getSize() + " bytes)");
}
final ByteArrayOutputStream baos = new ByteArrayOutputStream((int) flowFile.getSize());
session.read(flowFile, new InputStreamCallback() {
@Override
public void process(InputStream in) throws IOException {
try (BufferedInputStream input = new BufferedInputStream(in)) {
StreamUtils.copy(input, baos);
}
baos.close();
}
});
body = baos.toByteArray();
bodyLoaded = true;
}
}
return body;
}
@Override
public void setBody(byte[] body) {
synchronized (bodyLock) {
this.body = Arrays.copyOf(body, body.length);
bodyLoaded = true;
}
}
}

37
nifi-nar-bundles/nifi-flume-bundle/nifi-flume-processors/src/main/java/org/apache/nifi/processors/flume/util/FlowFileEventConstants.java
View File

@ -1,37 +0,0 @@
/*
* 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.flume.util;
public class FlowFileEventConstants {
// FlowFile#getEntryDate();
public static final String ENTRY_DATE_HEADER = "nifi.entry.date";
// FlowFile#getId();
public static final String ID_HEADER = "nifi.id";
// FlowFile#getLastQueueDate();
public static final String LAST_QUEUE_DATE_HEADER = "nifi.last.queue.date";
// FlowFile#getLineageStartDate();
public static final String LINEAGE_START_DATE_HEADER = "nifi.lineage.start.date";
// FlowFile#getSize();
public static final String SIZE_HEADER = "nifi.size";
}

16
nifi-nar-bundles/nifi-flume-bundle/nifi-flume-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
View File

@ -1,16 +0,0 @@
# 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.
org.apache.nifi.processors.flume.ExecuteFlumeSource
org.apache.nifi.processors.flume.ExecuteFlumeSink

155
nifi-nar-bundles/nifi-flume-bundle/nifi-flume-processors/src/main/resources/docs/org.apache.nifi.processors.flume.ExecuteFlumeSink/additionalDetails.html
View File

@ -1,155 +0,0 @@
<!DOCTYPE html>
<html lang="en">
<!--
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.
-->
<head>
<meta charset="utf-8" />
<title>ExecuteFlumeSink</title>
<link rel="stylesheet" href="../../../../../css/component-usage.css" type="text/css" />
</head>
<body>
<h2>Data Model</h2>
<p>
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:
</p>
<table id="headers">
<tr>
<th>Flume Event Header</th>
<th>FlowFile Attribute</th>
</tr>
<tr>
<td>nifi.entry.date</td>
<td>FlowFile#getEntryDate()</td>
</tr>
<tr>
<td>nifi.id</td>
<td>FlowFile#getId()</td>
</tr>
<tr>
<td>nifi.last.queue.date</td>
<td>FlowFile#getLastQueueDate()</td>
</tr>
<tr>
<td>nifi.lineage.start.date</td>
<td>FlowFile#getLineageStartDate()</td>
</tr>
<tr>
<td>nifi.size</td>
<td>FlowFile#getSize()</td>
</tr>
</table>
<h2>Warning</h2>
<p>
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.
</p>
<h2>Configuration Details</h2>
<p>
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
<a href="http://flume.apache.org/FlumeUserGuide.html#flume-sinks">User Guide</a>.
Configuring the Flume sink is a four step process:
</p>
<ol>
<li>Set the Sink Type property to a valid Flume sink type.</li>
<li>
Set the Agent Name property to the name of the agent in your
Flume configuration. This is the prefix of the properties in the Flume
configuration file. Example: <code>tier1</code>
</li>
<li>
Set the Sink Name property to the name of the sink in your Flume
configuration. If Agent Name is <code>tier1</code>, then the Sink Name
is the value of the <code>tier1.sinks</code> property. Example: <code>sink-1</code>
</li>
<li>
Copy and paste the configuration for the sink from your Flume configuration
file into the Flume Configuration property. Assuming you're using
the same Agent Name and Sink Name as in the examples above, this will be all
of the properties that start with <code>tier1.sinks.sink-1</code>.
Do not copy the <code>tier1.sinks.sink-1.type</code> or
<code>tier1.sinks.sink-1.channel</code> properties.
</li>
</ol>
<h2>Usage Example</h2>
<p>
Assuming you had the following existing Flume configuration file:
</p>
<pre>
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</pre>
<p>
Then you'd configure the ExecuteFlumeSink as follows:
</p>
<table id="example">
<tr>
<th>Property</th>
<th>Value</th>
</tr>
<tr>
<td>Sink Type</td>
<td>hdfs</td>
</tr>
<tr>
<td>Agent Name</td>
<td>a1</td>
</tr>
<tr>
<td>Sink Name</td>
<td>k1</td>
</tr>
<tr>
<td>Flume Configuration</td>
<td>
<code>
a1.sinks.k1.hdfs.path = /flume/events/%y-%m-%d/%H%M/%S<br>
a1.sinks.k1.hdfs.filePrefix = events-<br>
a1.sinks.k1.hdfs.round = true<br>
a1.sinks.k1.hdfs.roundValue = 10<br>
a1.sinks.k1.hdfs.roundUnit = minute
</code>
</td>
</tr>
</table>
</body>
</html>

114
nifi-nar-bundles/nifi-flume-bundle/nifi-flume-processors/src/main/resources/docs/org.apache.nifi.processors.flume.ExecuteFlumeSource/additionalDetails.html
View File

@ -1,114 +0,0 @@
<!DOCTYPE html>
<html lang="en">
<!--
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.
-->
<head>
<meta charset="utf-8" />
<title>ExecuteFlumeSource</title>
<link rel="stylesheet" href="../../../../../css/component-usage.css" type="text/css" />
</head>
<body>
<h2>Data Model</h2>
<p>
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.
</p>
<h2>Configuration Details</h2>
<p>
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
<a href="http://flume.apache.org/FlumeUserGuide.html#flume-sources">User Guide</a>.
Configuring the Flume source is a four step process:
</p>
<ol>
<li>Set the Source Type property to a valid Flume source type.</li>
<li>
Set the Agent Name property to the name of the agent in your
Flume configuration. This is the prefix of the properties in the Flume
configuration file. Example: <code>tier1</code>
</li>
<li>
Set the Source Name property to the name of the source in your Flume
configuration. If Agent Name is <code>tier1</code>, then the Source Name
is the value of the <code>tier1.sources</code> property. Example: <code>src-1</code>
</li>
<li>
Copy and paste the configuration for the source from your Flume configuration
file into the Flume Configuration property. Assuming you're using
the same Agent Name and Source Name as in the examples above, this will be all
of the properties that start with <code>tier1.sources.src-1</code>.
Do not copy the <code>tier1.sources.src-1.type</code> or
<code>tier1.sources.src-1.channel</code> properties.
</li>
</ol>
<h2>Usage Example</h2>
<p>
Assuming you had the following existing Flume configuration file:
</p>
<pre>
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</pre>
<p>
Then you'd configure the ExecuteFlumeSource as follows:
</p>
<table id="example">
<tr>
<th>Property</th>
<th>Value</th>
</tr>
<tr>
<td>Source Type</td>
<td>multiport_syslogtcp</td>
</tr>
<tr>
<td>Agent Name</td>
<td>a1</td>
</tr>
<tr>
<td>Source Name</td>
<td>r1</td>
</tr>
<tr>
<td>Flume Configuration</td>
<td>
<code>
a1.sources.r1.host = 0.0.0.0<br>
a1.sources.r1.ports = 10001 10002 10003<br>
a1.sources.r1.portHeader = port
</code>
</td>
</tr>
</table>
</body>
</html>

168
nifi-nar-bundles/nifi-flume-bundle/nifi-flume-processors/src/test/java/org/apache/nifi/processors/flume/ExecuteFlumeSinkTest.java
View File

@ -1,168 +0,0 @@
/*
* 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.flume;
import org.apache.commons.io.filefilter.HiddenFileFilter;
import org.apache.flume.sink.NullSink;
import org.apache.flume.source.AvroSource;
import org.apache.nifi.components.ValidationResult;
import org.apache.nifi.flowfile.attributes.CoreAttributes;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.util.MockProcessContext;
import org.apache.nifi.util.TestRunner;
import org.apache.nifi.util.TestRunners;
import org.apache.nifi.util.file.FileUtils;
import org.apache.nifi.util.security.MessageDigestUtils;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.FilenameFilter;
import java.io.IOException;
import java.io.InputStream;
import java.nio.file.Path;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
public class ExecuteFlumeSinkTest {
private static final Logger logger =
LoggerFactory.getLogger(ExecuteFlumeSinkTest.class);
@Test
public void testValidators() {
TestRunner runner = TestRunners.newTestRunner(ExecuteFlumeSink.class);
Collection<ValidationResult> results;
ProcessContext pc;
results = new HashSet<>();
runner.enqueue(new byte[0]);
pc = runner.getProcessContext();
if (pc instanceof MockProcessContext) {
results = ((MockProcessContext) pc).validate();
}
assertEquals(1, results.size());
for (ValidationResult vr : results) {
logger.debug(vr.toString());
assertTrue(vr.toString().contains("is invalid because Sink Type is required"));
}
// non-existent class
results = new HashSet<>();
runner.setProperty(ExecuteFlumeSink.SINK_TYPE, "invalid.class.name");
runner.enqueue(new byte[0]);
pc = runner.getProcessContext();
if (pc instanceof MockProcessContext) {
results = ((MockProcessContext) pc).validate();
}
assertEquals(1, results.size());
for (ValidationResult vr : results) {
logger.debug(vr.toString());
assertTrue(vr.toString().contains("is invalid because unable to load sink"));
}
// class doesn't implement Sink
results = new HashSet<>();
runner.setProperty(ExecuteFlumeSink.SINK_TYPE, AvroSource.class.getName());
runner.enqueue(new byte[0]);
pc = runner.getProcessContext();
if (pc instanceof MockProcessContext) {
results = ((MockProcessContext) pc).validate();
}
assertEquals(1, results.size());
for (ValidationResult vr : results) {
logger.debug(vr.toString());
assertTrue(vr.toString().contains("is invalid because unable to create sink"));
}
results = new HashSet<>();
runner.setProperty(ExecuteFlumeSink.SINK_TYPE, NullSink.class.getName());
runner.enqueue(new byte[0]);
pc = runner.getProcessContext();
if (pc instanceof MockProcessContext) {
results = ((MockProcessContext) pc).validate();
}
assertEquals(0, results.size());
}
@Test
public void testNullSink() throws IOException {
TestRunner runner = TestRunners.newTestRunner(ExecuteFlumeSink.class);
runner.setProperty(ExecuteFlumeSink.SINK_TYPE, NullSink.class.getName());
try (InputStream inputStream = getClass().getResourceAsStream("/testdata/records.txt")) {
Map<String, String> attributes = new HashMap<>();
attributes.put(CoreAttributes.FILENAME.key(), "records.txt");
runner.enqueue(inputStream, attributes);
runner.run();
}
}
@Test
public void testBatchSize() throws IOException {
TestRunner runner = TestRunners.newTestRunner(ExecuteFlumeSink.class);
runner.setProperty(ExecuteFlumeSink.SINK_TYPE, NullSink.class.getName());
runner.setProperty(ExecuteFlumeSink.FLUME_CONFIG,
"tier1.sinks.sink-1.batchSize = 1000\n");
for (int i = 0; i < 100000; i++) {
runner.enqueue(String.valueOf(i).getBytes());
}
runner.run(100);
}
@Test
@Disabled("Does not work on Windows")
public void testHdfsSink(@TempDir Path temp) throws IOException {
File destDir = temp.resolve("hdfs").toFile();
destDir.mkdirs();
TestRunner runner = TestRunners.newTestRunner(ExecuteFlumeSink.class);
runner.setProperty(ExecuteFlumeSink.SINK_TYPE, "hdfs");
runner.setProperty(ExecuteFlumeSink.FLUME_CONFIG,
"tier1.sinks.sink-1.hdfs.path = " + destDir.toURI().toString() + "\n" +
"tier1.sinks.sink-1.hdfs.fileType = DataStream\n" +
"tier1.sinks.sink-1.hdfs.serializer = TEXT\n" +
"tier1.sinks.sink-1.serializer.appendNewline = false"
);
try (InputStream inputStream = getClass().getResourceAsStream("/testdata/records.txt")) {
Map<String, String> attributes = new HashMap<>();
attributes.put(CoreAttributes.FILENAME.key(), "records.txt");
runner.enqueue(inputStream, attributes);
runner.run();
}
File[] files = destDir.listFiles((FilenameFilter)HiddenFileFilter.VISIBLE);
assertEquals(1, files.length, "Unexpected number of destination files.");
File dst = files[0];
byte[] expectedDigest;
try (InputStream resourceStream = getClass().getResourceAsStream("/testdata/records.txt")) {
expectedDigest = MessageDigestUtils.getDigest(resourceStream);
}
byte[] actualDigest = FileUtils.computeDigest(dst);
assertArrayEquals(expectedDigest, actualDigest, "Destination file doesn't match source data");
}
}

151
nifi-nar-bundles/nifi-flume-bundle/nifi-flume-processors/src/test/java/org/apache/nifi/processors/flume/ExecuteFlumeSourceTest.java
View File

@ -1,151 +0,0 @@
/*
* 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.flume;
import org.apache.flume.sink.NullSink;
import org.apache.flume.source.AvroSource;
import org.apache.nifi.components.ValidationResult;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.util.MockFlowFile;
import org.apache.nifi.util.MockProcessContext;
import org.apache.nifi.util.TestRunner;
import org.apache.nifi.util.TestRunners;
import org.apache.nifi.util.file.FileUtils;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.IOException;
import java.nio.file.Path;
import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.concurrent.TimeUnit;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
public class ExecuteFlumeSourceTest {
private static final Logger logger = LoggerFactory.getLogger(ExecuteFlumeSourceTest.class);
@Test
public void testValidators() {
TestRunner runner = TestRunners.newTestRunner(ExecuteFlumeSource.class);
Collection<ValidationResult> results;
ProcessContext pc;
results = new HashSet<>();
runner.enqueue(new byte[0]);
pc = runner.getProcessContext();
if (pc instanceof MockProcessContext) {
results = ((MockProcessContext) pc).validate();
}
assertEquals(1, results.size());
for (ValidationResult vr : results) {
logger.debug(vr.toString());
assertTrue(vr.toString().contains("is invalid because Source Type is required"));
}
// non-existent class
results = new HashSet<>();
runner.setProperty(ExecuteFlumeSource.SOURCE_TYPE, "invalid.class.name");
runner.enqueue(new byte[0]);
pc = runner.getProcessContext();
if (pc instanceof MockProcessContext) {
results = ((MockProcessContext) pc).validate();
}
assertEquals(1, results.size());
for (ValidationResult vr : results) {
logger.debug(vr.toString());
assertTrue(vr.toString().contains("is invalid because unable to load source"));
}
// class doesn't implement Source
results = new HashSet<>();
runner.setProperty(ExecuteFlumeSource.SOURCE_TYPE, NullSink.class.getName());
runner.enqueue(new byte[0]);
pc = runner.getProcessContext();
if (pc instanceof MockProcessContext) {
results = ((MockProcessContext) pc).validate();
}
assertEquals(1, results.size());
for (ValidationResult vr : results) {
logger.debug(vr.toString());
assertTrue(vr.toString().contains("is invalid because unable to create source"));
}
results = new HashSet<>();
runner.setProperty(ExecuteFlumeSource.SOURCE_TYPE, AvroSource.class.getName());
runner.enqueue(new byte[0]);
pc = runner.getProcessContext();
if (pc instanceof MockProcessContext) {
results = ((MockProcessContext) pc).validate();
}
assertEquals(0, results.size());
}
@Test
public void testSequenceSource() {
TestRunner runner = TestRunners.newTestRunner(ExecuteFlumeSource.class);
runner.setProperty(ExecuteFlumeSource.SOURCE_TYPE, "seq");
runner.run();
List<MockFlowFile> flowFiles = runner.getFlowFilesForRelationship(ExecuteFlumeSource.SUCCESS);
assertEquals(1, flowFiles.size());
for (MockFlowFile flowFile : flowFiles) {
logger.debug(flowFile.toString());
assertEquals(1, flowFile.getSize());
}
}
@Test
@Disabled("Does not work on Windows")
public void testSourceWithConfig(@TempDir Path temp) throws IOException {
File spoolDirectory = temp.resolve("spooldir").toFile();
spoolDirectory.mkdirs();
File dst = new File(spoolDirectory, "records.txt");
FileUtils.copyFile(getClass().getResourceAsStream("/testdata/records.txt"), dst, true, false);
TestRunner runner = TestRunners.newTestRunner(ExecuteFlumeSource.class);
runner.setProperty(ExecuteFlumeSource.SOURCE_TYPE, "spooldir");
runner.setProperty(ExecuteFlumeSink.FLUME_CONFIG,
"tier1.sources.src-1.spoolDir = " + spoolDirectory.getAbsolutePath());
runner.run(1, false, true);
// Because the spool directory source is an event driven source, it may take some time for flow files to get
// produced. I'm willing to wait up to 5 seconds, but will bail out early if possible. If it takes longer than
// that then there is likely a bug.
int numWaits = 10;
while (runner.getFlowFilesForRelationship(ExecuteFlumeSource.SUCCESS).size() < 4 && --numWaits > 0) {
try {
TimeUnit.MILLISECONDS.sleep(500);
} catch (InterruptedException ex) {
logger.warn("Sleep interrupted");
}
}
runner.shutdown();
runner.assertTransferCount(ExecuteFlumeSource.SUCCESS, 4);
int i = 1;
for (MockFlowFile flowFile : runner.getFlowFilesForRelationship(ExecuteFlumeSource.SUCCESS)) {
flowFile.assertContentEquals("record " + i);
i++;
}
}
}

25
nifi-nar-bundles/nifi-flume-bundle/nifi-flume-processors/src/test/resources/core-site-broken.xml
View File

@ -1,25 +0,0 @@
<?xml version="1.0"?>
<!--
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.
-->
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
<!-- Put site-specific property overrides in this file. -->
<configuration>
<property>
<name>fs.default.name</name>
<value>hdfs://localhost:65535</value>
</property>
</configuration>

30
nifi-nar-bundles/nifi-flume-bundle/nifi-flume-processors/src/test/resources/core-site.xml
View File

@ -1,30 +0,0 @@
<?xml version="1.0"?>
<!--
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.
-->
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
<!-- Put site-specific property overrides in this file. -->
<configuration>
<property>
<name>fs.defaultFS</name>
<!--
Hadoop doesn't support a chroot style operation for the
local filesystem so there's no benefit to setting this
to a directory other than '/'
-->
<value>file:///</value>
</property>
</configuration>

21
nifi-nar-bundles/nifi-flume-bundle/nifi-flume-processors/src/test/resources/simplelogger.properties
View File

@ -1,21 +0,0 @@
# 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.
org.slf4j.simpleLogger.logFile=System.out
org.slf4j.simpleLogger.defaultLogLevel=info
org.slf4j.simpleLogger.showDateTime=true
org.slf4j.simpleLogger.dateTimeFormat=yyyy-MM-dd HH:mm:ss.SSS
org.slf4j.simpleLogger.levelInBrackets=true
org.slf4j.simpleLogger.log.org.apache.nifi.processors.flume=debug
org.slf4j.simpleLogger.log.org.apache.flume=debug

4
nifi-nar-bundles/nifi-flume-bundle/nifi-flume-processors/src/test/resources/testdata/records.txt vendored
View File

@ -1,4 +0,0 @@
record 1
record 2
record 3
record 4

51
nifi-nar-bundles/nifi-flume-bundle/pom.xml
View File

@ -1,51 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
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.
-->
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-nar-bundles</artifactId>
<version>2.0.0-SNAPSHOT</version>
</parent>
<artifactId>nifi-flume-bundle</artifactId>
<packaging>pom</packaging>
<description>A bundle of processors that run Flume sources/sinks</description>
<modules>
<module>nifi-flume-processors</module>
<module>nifi-flume-nar</module>
</modules>
<dependencyManagement>
<dependencies>
<dependency>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-flume-processors</artifactId>
<version>2.0.0-SNAPSHOT</version>
</dependency>
<!-- Override Netty 3 -->
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty</artifactId>
<version>${netty.3.version}</version>
</dependency>
<!-- Override Tomcat Embed Core 8.5.46 from Flume 1.10.0 -->
<dependency>
<groupId>org.apache.tomcat.embed</groupId>
<artifactId>tomcat-embed-core</artifactId>
<version>8.5.85</version>
</dependency>
</dependencies>
</dependencyManagement>
</project>

1
nifi-nar-bundles/pom.xml
View File

@ -45,7 +45,6 @@
<module>nifi-ccda-bundle</module>
<module>nifi-language-translation-bundle</module>
<module>nifi-mongodb-bundle</module>
<module>nifi-flume-bundle</module>
<module>nifi-hbase-bundle</module>
<module>nifi-ambari-bundle</module>
<module>nifi-asana-bundle</module>