You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@logging.apache.org by rg...@apache.org on 2026/09/01 20:47:00 UTC
(logging-flume-mongodb) 01/01: Initial commit
This is an automated email from the ASF dual-hosted git repository.
rgoers pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/logging-flume-mongodb.git
commit 31aadeb1e0b4f4158f8e93b8a6490198c140c5be
Author: Ralph Goers <rg...@apache.org>
AuthorDate: Tue Sep 1 13:46:45 2026 -0700
Initial commit
---
.asf.yaml | 98 +++++++
CHANGELOG | 7 +
LICENSE.txt | 245 ++++++++++++++++++
NOTICE.txt | 5 +
README.md | 58 +++++
RELEASE-NOTES.txt | 27 ++
checkstyle-header.txt | 16 ++
findbugs-exclude-filter.xml | 31 +++
flume-mongodb-sink/.flattened-pom.xml | 132 ++++++++++
flume-mongodb-sink/pom.xml | 143 +++++++++++
.../flume/sink/mongodb/DefaultMongoDbWriter.java | 52 ++++
.../org/apache/flume/sink/mongodb/MongoDbSink.java | 285 +++++++++++++++++++++
.../flume/sink/mongodb/MongoDbSinkConstants.java | 71 +++++
.../apache/flume/sink/mongodb/MongoDbWriter.java | 43 ++++
.../apache/flume/sink/mongodb/TestMongoDbSink.java | 261 +++++++++++++++++++
flume-mongodb-sink/src/test/resources/log4j2.xml | 71 +++++
pom.xml | 229 +++++++++++++++++
17 files changed, 1774 insertions(+)
diff --git a/.asf.yaml b/.asf.yaml
new file mode 100644
index 0000000..7a2c143
--- /dev/null
+++ b/.asf.yaml
@@ -0,0 +1,98 @@
+#
+# 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.
+#
+
+# `.asf.yaml` is a branch-specific YAML configuration file for Git repositories to control features such as notifications, GitHub settings, etc.
+# See its documentation for details: https://cwiki.apache.org/confluence/display/INFRA/Git+-+.asf.yaml+features
+
+notifications:
+ # GitHub already provides notifications for PRs and issues.
+ # Please don't duplicate that noise here!
+ commits: commits@logging.apache.org
+ jira_options: link label
+github:
+ description: "Apache Flume is a distributed, reliable, and available service for efficiently collecting, aggregating, and moving large amounts of log-like data"
+ homepage: https://flume.apache.org/
+ autolink_jira:
+ - FLUME
+ labels:
+ - apache
+ - api
+ - java
+ - jvm
+ - library
+ - flume
+
+ # Pull Request settings:
+ # https://github.com/apache/infrastructure-asfyaml#pull-request-settings
+ pull_requests:
+ # allow auto-merge
+ allow_auto_merge: true
+ # enable updating head branches of pull requests
+ allow_update_branch: true
+ # auto-delete head branches after being merged
+ del_branch_on_merge: true
+
+ # Enforce squashing while merging PRs.
+ # Otherwise, the git log gets polluted severely.
+ enabled_merge_buttons:
+ squash: true
+ merge: false
+ rebase: false
+
+ features:
+ issues: true
+
+ # Disable protected branches, since we use rulesets now
+ protected_branches: ~
+
+ rulesets:
+ - name: "Branch protection"
+ type: branch
+ branches:
+ includes:
+ - "~DEFAULT_BRANCH"
+ # All reviews must be addressed before merging
+ required_conversation_resolution: true
+ # Require checks to pass before merging
+ required_status_checks:
+ # The GitHub Actions app
+ - app_slug: github-actions
+ name: "build (ubuntu-latest)"
+ # The GitHub Advanced Security app
+ - app_slug: github-advanced-security
+ name: "CodeQL"
+ # At least one positive review must be present
+ required_pull_request_reviews:
+ required_approving_review_count: 0
+ # Use raw rules, until a convenience notation for `restrict_update` is introduced.
+ # See: https://github.com/apache/infrastructure-asfyaml/issues/96
+ #
+ # The raw rules need to follow the syntax given in:
+ # https://docs.github.com/en/rest/repos/rules?apiVersion=2026-03-10#update-a-repository-ruleset
+ - name: "Tag protection"
+ target: tag
+ enforcement: active
+ bypass_actors: []
+ conditions:
+ ref_name:
+ include:
+ - "refs/tags/rel/*"
+ exclude: []
+ rules:
+ - type: deletion
+ - type: non_fast_forward
+ - type: update
diff --git a/CHANGELOG b/CHANGELOG
new file mode 100644
index 0000000..48cec67
--- /dev/null
+++ b/CHANGELOG
@@ -0,0 +1,7 @@
+Release Notes - Flume MongoDb - Version 2.0.0
+
+** Improvement
+ * Initial version.
+
+
+
diff --git a/LICENSE.txt b/LICENSE.txt
new file mode 100644
index 0000000..9fa7156
--- /dev/null
+++ b/LICENSE.txt
@@ -0,0 +1,245 @@
+
+ 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.
+
+====
+
+The following files are included under the 2-Clause BSD License
+
+flume-ng-sinks/flume-ng-morphline-solr-sink/src/test/resources/solr/collection1/conf/lang/stopwords_ar.txt
+flume-ng-sinks/flume-ng-morphline-solr-sink/src/test/resources/solr/collection1/conf/lang/stopwords_bg.txt
+flume-ng-sinks/flume-ng-morphline-solr-sink/src/test/resources/solr/collection1/conf/lang/stopwords_da.txt
+flume-ng-sinks/flume-ng-morphline-solr-sink/src/test/resources/solr/collection1/conf/lang/stopwords_de.txt
+flume-ng-sinks/flume-ng-morphline-solr-sink/src/test/resources/solr/collection1/conf/lang/stopwords_es.txt
+flume-ng-sinks/flume-ng-morphline-solr-sink/src/test/resources/solr/collection1/conf/lang/stopwords_fa.txt
+flume-ng-sinks/flume-ng-morphline-solr-sink/src/test/resources/solr/collection1/conf/lang/stopwords_fi.txt
+flume-ng-sinks/flume-ng-morphline-solr-sink/src/test/resources/solr/collection1/conf/lang/stopwords_fr.txt
+flume-ng-sinks/flume-ng-morphline-solr-sink/src/test/resources/solr/collection1/conf/lang/stopwords_hi.txt
+flume-ng-sinks/flume-ng-morphline-solr-sink/src/test/resources/solr/collection1/conf/lang/stopwords_hu.txt
+flume-ng-sinks/flume-ng-morphline-solr-sink/src/test/resources/solr/collection1/conf/lang/stopwords_it.txt
+flume-ng-sinks/flume-ng-morphline-solr-sink/src/test/resources/solr/collection1/conf/lang/stopwords_nl.txt
+flume-ng-sinks/flume-ng-morphline-solr-sink/src/test/resources/solr/collection1/conf/lang/stopwords_no.txt
+flume-ng-sinks/flume-ng-morphline-solr-sink/src/test/resources/solr/collection1/conf/lang/stopwords_pt.txt
+flume-ng-sinks/flume-ng-morphline-solr-sink/src/test/resources/solr/collection1/conf/lang/stopwords_ro.txt
+flume-ng-sinks/flume-ng-morphline-solr-sink/src/test/resources/solr/collection1/conf/lang/stopwords_ru.txt
+flume-ng-sinks/flume-ng-morphline-solr-sink/src/test/resources/solr/collection1/conf/lang/stopwords_sv.txt
+
+Redistribution and use in source and binary forms, with or without modification,
+are permitted provided that the following conditions are met:
+
+1. Redistributions of source code must retain the above copyright notice, this
+list of conditions and the following disclaimer.
+
+2. 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.
+
+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.
diff --git a/NOTICE.txt b/NOTICE.txt
new file mode 100644
index 0000000..a2c2ff9
--- /dev/null
+++ b/NOTICE.txt
@@ -0,0 +1,5 @@
+Apache Flume Spring Boot
+Copyright 2022-2023 The Apache Software Foundation
+
+This product includes software developed at
+The Apache Software Foundation (http://www.apache.org/).
diff --git a/README.md b/README.md
new file mode 100644
index 0000000..52a929a
--- /dev/null
+++ b/README.md
@@ -0,0 +1,58 @@
+<!--
+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.
+-->
+
+# Welcome to Apache Flume MongoDB!
+
+Apache Flume is a distributed, reliable, and available service for efficiently
+collecting, aggregating, and moving large amounts of event data. It has a simple
+and flexible architecture based on streaming data flows. It is robust and fault
+tolerant with tunable reliability mechanisms and many failover and recovery
+mechanisms. The system is centrally managed and allows for intelligent dynamic
+management. It uses a simple extensible data model that allows for online
+analytic application.
+
+The Apache Flume MongoDB provides a MongoDb Sink.
+
+Apache Flume MongoDb is open-sourced under the Apache Software Foundation License v2.0.
+
+## Documentation
+
+Documentation is included in the binary distribution under the docs directory.
+In source form, it can be found in the flume-ng-doc directory.
+
+The Flume 2.x guide and FAQ are available here:
+
+* https://cwiki.apache.org/FLUME
+* https://cwiki.apache.org/confluence/display/FLUME/Getting+Started
+
+## Contact us!
+
+* Mailing lists: https://cwiki.apache.org/confluence/display/FLUME/Mailing+Lists
+* Slack channel #flume on https://the-asf.slack.com/
+
+Bug and Issue tracker.
+
+* https://github.com/apache/logging-flume-mongodb/issues
+
+## Compiling Flume MongoDb
+
+Compiling Flume MongoDb requires the following tools:
+
+* Oracle Java JDK 11
+* Apache Maven 3.x
diff --git a/RELEASE-NOTES.txt b/RELEASE-NOTES.txt
new file mode 100644
index 0000000..0cbbad9
--- /dev/null
+++ b/RELEASE-NOTES.txt
@@ -0,0 +1,27 @@
+Apache Flume Kafka 2.0.0
+
+CONTENTS
+1. What is Apache Flume Kafka
+2. Major changes in this Release
+3. How to Get Involved
+4. How to Report Issues
+
+1. What is Apache Flume Kafka
+Flume is a distributed, reliable, and available service for
+efficiently collecting, aggregating, and moving large amounts of event
+data. Flume Kafka provides a Channel, Source, and Sink for publishing
+and consuming events to and from Kafka.
+
+2. Major changes in this Release
+For a detailed list of changes, please see the CHANGELOG file included
+in this distribution.
+
+3. How to Get Involved
+The Apache Flume project really needs and appreciates any contributions,
+including documentation help, source code and feedback. If you are interested
+in contributing, please visit:
+https://cwiki.apache.org/confluence/display/FLUME/How+to+Contribute
+
+4. How to Report Issues
+The Apache Flume Spring Boot project uses GitHub issues for issue tracking. Please see
+https://github.com/apache/flume-spring-boot/issues
diff --git a/checkstyle-header.txt b/checkstyle-header.txt
new file mode 100644
index 0000000..4f33236
--- /dev/null
+++ b/checkstyle-header.txt
@@ -0,0 +1,16 @@
+/*
+ * 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.
+ */
diff --git a/findbugs-exclude-filter.xml b/findbugs-exclude-filter.xml
new file mode 100644
index 0000000..327be31
--- /dev/null
+++ b/findbugs-exclude-filter.xml
@@ -0,0 +1,31 @@
+<?xml version="1.0" encoding="iso-8859-1"?>
+<!--
+ 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.
+-->
+
+<!-- ===================================================================== -->
+<!-- $Id: findbugs-exclude-filter.xml 773234 2009-05-09 15:27:59Z rgoers $ -->
+<!-- ===================================================================== -->
+<FindBugsFilter>
+ <!-- Enable only high priority warnings -->
+ <Match>
+ <Priority value="2"/>
+ </Match>
+
+ <Match>
+ <Priority value="3"/>
+ </Match>
+</FindBugsFilter>
diff --git a/flume-mongodb-sink/.flattened-pom.xml b/flume-mongodb-sink/.flattened-pom.xml
new file mode 100644
index 0000000..26355e6
--- /dev/null
+++ b/flume-mongodb-sink/.flattened-pom.xml
@@ -0,0 +1,132 @@
+<?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 xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd" xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.flume</groupId>
+ <artifactId>flume-mongodb-parent</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ </parent>
+ <groupId>org.apache.flume</groupId>
+ <artifactId>flume-mongodb-sink</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ <name>Flume MongoDb Sink</name>
+ <url>https://logging.apache.org/flume/2.x/index.html/flume-mongodb-parent/flume-mongodb-sink</url>
+ <licenses>
+ <license>
+ <name>The Apache Software License, Version 2.0</name>
+ <url>http://www.apache.org/licenses/LICENSE-2.0.txt</url>
+ </license>
+ </licenses>
+ <properties>
+ <spotbugs.maxAllowedViolations>3</spotbugs.maxAllowedViolations>
+ <pmd.maxAllowedViolations>3</pmd.maxAllowedViolations>
+ <module.name>org.apache.flume.sink.mongodb</module.name>
+ </properties>
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.flume</groupId>
+ <artifactId>flume-ng-sdk</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flume</groupId>
+ <artifactId>flume-ng-core</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flume</groupId>
+ <artifactId>flume-ng-configuration</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.mongodb</groupId>
+ <artifactId>mongodb-driver-sync</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>io.dropwizard.metrics</groupId>
+ <artifactId>metrics-core</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-api</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-core</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-all</artifactId>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+ <build>
+ <plugins>
+ <plugin>
+ <artifactId>maven-compiler-plugin</artifactId>
+ <version>3.13.0</version>
+ <configuration>
+ <compilerArgs>
+ <arg>-J--add-exports=jdk.compiler/com.sun.tools.javac.api=ALL-UNNAMED</arg>
+ <arg>-J--add-exports=jdk.compiler/com.sun.tools.javac.file=ALL-UNNAMED</arg>
+ <arg>-J--add-exports=jdk.compiler/com.sun.tools.javac.main=ALL-UNNAMED</arg>
+ <arg>-J--add-exports=jdk.compiler/com.sun.tools.javac.model=ALL-UNNAMED</arg>
+ <arg>-J--add-exports=jdk.compiler/com.sun.tools.javac.parser=ALL-UNNAMED</arg>
+ <arg>-J--add-exports=jdk.compiler/com.sun.tools.javac.processing=ALL-UNNAMED</arg>
+ <arg>-J--add-exports=jdk.compiler/com.sun.tools.javac.tree=ALL-UNNAMED</arg>
+ <arg>-J--add-exports=jdk.compiler/com.sun.tools.javac.util=ALL-UNNAMED</arg>
+ <arg>-J--add-exports=jdk.compiler/com.sun.tools.javac.code=ALL-UNNAMED</arg>
+ <arg>-J--add-exports=jdk.compiler/com.sun.tools.javac.comp=ALL-UNNAMED</arg>
+ </compilerArgs>
+ </configuration>
+ </plugin>
+ <plugin>
+ <groupId>biz.aQute.bnd</groupId>
+ <artifactId>bnd-baseline-maven-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>check-api-compat</id>
+ <goals>
+ <goal>baseline</goal>
+ </goals>
+ </execution>
+ </executions>
+ <configuration>
+ <failOnMissing>false</failOnMissing>
+ <includeDistributionManagement>false</includeDistributionManagement>
+ <releaseversions>true</releaseversions>
+ </configuration>
+ </plugin>
+ <plugin>
+ <artifactId>maven-jar-plugin</artifactId>
+ <executions>
+ <execution>
+ <goals>
+ <goal>test-jar</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+</project>
diff --git a/flume-mongodb-sink/pom.xml b/flume-mongodb-sink/pom.xml
new file mode 100644
index 0000000..da3c4b8
--- /dev/null
+++ b/flume-mongodb-sink/pom.xml
@@ -0,0 +1,143 @@
+<?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 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.flume</groupId>
+ <artifactId>flume-mongodb-parent</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ <relativePath>../pom.xml</relativePath>
+ </parent>
+ <groupId>org.apache.flume</groupId>
+ <artifactId>flume-mongodb-sink</artifactId>
+ <name>Flume MongoDb Sink</name>
+
+ <properties>
+ <!-- TODO fix spotbugs/pmd violations -->
+ <spotbugs.maxAllowedViolations>3</spotbugs.maxAllowedViolations>
+ <pmd.maxAllowedViolations>3</pmd.maxAllowedViolations>
+ <module.name>org.apache.flume.sink.mongodb</module.name>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.flume</groupId>
+ <artifactId>flume-ng-sdk</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flume</groupId>
+ <artifactId>flume-ng-core</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flume</groupId>
+ <artifactId>flume-ng-configuration</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.mongodb</groupId>
+ <artifactId>mongodb-driver-sync</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>io.dropwizard.metrics</groupId>
+ <artifactId>metrics-core</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-api</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-core</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-all</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-compiler-plugin</artifactId>
+ <version>3.13.0</version>
+ <!-- Use a recent version -->
+ <configuration>
+ <compilerArgs>
+ <arg>-J--add-exports=jdk.compiler/com.sun.tools.javac.api=ALL-UNNAMED</arg>
+ <arg>-J--add-exports=jdk.compiler/com.sun.tools.javac.file=ALL-UNNAMED</arg>
+ <arg>-J--add-exports=jdk.compiler/com.sun.tools.javac.main=ALL-UNNAMED</arg>
+ <arg>-J--add-exports=jdk.compiler/com.sun.tools.javac.model=ALL-UNNAMED</arg>
+ <arg>-J--add-exports=jdk.compiler/com.sun.tools.javac.parser=ALL-UNNAMED</arg>
+ <arg>-J--add-exports=jdk.compiler/com.sun.tools.javac.processing=ALL-UNNAMED</arg>
+ <arg>-J--add-exports=jdk.compiler/com.sun.tools.javac.tree=ALL-UNNAMED</arg>
+ <arg>-J--add-exports=jdk.compiler/com.sun.tools.javac.util=ALL-UNNAMED</arg>
+ <arg>-J--add-exports=jdk.compiler/com.sun.tools.javac.code=ALL-UNNAMED</arg>
+ <arg>-J--add-exports=jdk.compiler/com.sun.tools.javac.comp=ALL-UNNAMED</arg>
+ </compilerArgs>
+ </configuration>
+ </plugin>
+ <plugin>
+ <groupId>biz.aQute.bnd</groupId>
+ <artifactId>bnd-baseline-maven-plugin</artifactId>
+ <configuration>
+ <failOnMissing>false</failOnMissing>
+ <!-- Do not use the <distributionManagement&rt; element to download releases -->
+ <includeDistributionManagement>false</includeDistributionManagement>
+ <!-- Do not check against alpha releases -->
+ <releaseversions>true</releaseversions>
+ </configuration>
+ <executions>
+ <execution>
+ <id>check-api-compat</id>
+ <goals>
+ <goal>baseline</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-jar-plugin</artifactId>
+ <executions>
+ <execution>
+ <goals>
+ <goal>test-jar</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+
+</project>
diff --git a/flume-mongodb-sink/src/main/java/org/apache/flume/sink/mongodb/DefaultMongoDbWriter.java b/flume-mongodb-sink/src/main/java/org/apache/flume/sink/mongodb/DefaultMongoDbWriter.java
new file mode 100644
index 0000000..d039717
--- /dev/null
+++ b/flume-mongodb-sink/src/main/java/org/apache/flume/sink/mongodb/DefaultMongoDbWriter.java
@@ -0,0 +1,52 @@
+/*
+ * 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.flume.sink.mongodb;
+
+import com.mongodb.WriteConcern;
+import com.mongodb.client.MongoCollection;
+import com.mongodb.client.MongoDatabase;
+import java.util.List;
+import org.bson.Document;
+
+/**
+ * {@link MongoDbWriter} implementation backed by a real MongoDB
+ * {@link MongoDatabase} connection.
+ */
+public class DefaultMongoDbWriter implements MongoDbWriter {
+
+ private final MongoDatabase mongoDatabase;
+ private final WriteConcern writeConcern;
+
+ public DefaultMongoDbWriter(MongoDatabase mongoDatabase, WriteConcern writeConcern) {
+ this.mongoDatabase = mongoDatabase;
+ this.writeConcern = writeConcern;
+ }
+
+ @Override
+ public void write(String collectionName, List<Document> documents) {
+ MongoCollection<Document> collection = mongoDatabase.getCollection(collectionName);
+ if (writeConcern != null) {
+ collection = collection.withWriteConcern(writeConcern);
+ }
+ collection.insertMany(documents);
+ }
+
+ @Override
+ public void close() {
+ // The underlying MongoClient is owned and closed by MongoDbSink.
+ }
+}
diff --git a/flume-mongodb-sink/src/main/java/org/apache/flume/sink/mongodb/MongoDbSink.java b/flume-mongodb-sink/src/main/java/org/apache/flume/sink/mongodb/MongoDbSink.java
new file mode 100644
index 0000000..9c32289
--- /dev/null
+++ b/flume-mongodb-sink/src/main/java/org/apache/flume/sink/mongodb/MongoDbSink.java
@@ -0,0 +1,285 @@
+/*
+ * 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.flume.sink.mongodb;
+
+import com.mongodb.MongoException;
+import com.mongodb.WriteConcern;
+import com.mongodb.client.MongoClient;
+import com.mongodb.client.MongoClients;
+import com.mongodb.client.MongoDatabase;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.flume.Channel;
+import org.apache.flume.Context;
+import org.apache.flume.Event;
+import org.apache.flume.EventDeliveryException;
+import org.apache.flume.Transaction;
+import org.apache.flume.conf.BatchSizeSupported;
+import org.apache.flume.conf.Configurable;
+import org.apache.flume.conf.ConfigurationException;
+import org.apache.flume.instrumentation.SinkCounter;
+import org.apache.flume.sink.AbstractSink;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.bson.Document;
+
+/**
+ * A Flume Sink that writes events to MongoDB.
+ * <p/>
+ * Each batch of events read from the channel is converted into MongoDB
+ * {@link Document}s and inserted into a collection. The target collection
+ * for a given event may be determined dynamically by reading a Flume event
+ * header and mapping its value onto one of a set of predefined collection
+ * names, allowing a single sink to fan out events to different collections
+ * based on the header value.
+ * <p/>
+ * Mandatory properties are:
+ * mongodb.uri -- the MongoDB connection string
+ * mongodb.database -- the database to write to
+ * mongodb.collection -- the default/fallback collection to write to
+ * <p/>
+ * Optional properties:
+ * mongodb.collectionHeader -- name of the event header used to select the
+ * target collection
+ * mongodb.collectionMap.<headerValue> -- maps a header value to an
+ * actual collection name
+ * mongodb.collectionMapFallback -- whether to fall back to the default
+ * collection when the header value has no mapping (default true)
+ * mongodb.includeHeaders -- whether to include the Flume event headers in
+ * the resulting document (default false)
+ * mongodb.writeConcern -- the write concern to use
+ * batchSize -- how many events to process in one batch (default 100)
+ */
+public class MongoDbSink extends AbstractSink implements Configurable, BatchSizeSupported {
+
+ private static final Logger logger = LogManager.getLogger(MongoDbSink.class);
+
+ private String connectionUri;
+ private String databaseName;
+ private String defaultCollection;
+ private String collectionHeader;
+ private Map<String, String> collectionMap;
+ private boolean collectionMapFallback;
+ private boolean includeHeaders;
+ private int batchSize;
+ private WriteConcern writeConcern;
+
+ private MongoClient mongoClient;
+ private MongoDbWriter writer;
+ private SinkCounter counter;
+
+ // For testing
+ public String getDatabaseName() {
+ return databaseName;
+ }
+
+ public String getDefaultCollection() {
+ return defaultCollection;
+ }
+
+ @Override
+ public long getBatchSize() {
+ return batchSize;
+ }
+
+ @Override
+ public Status process() throws EventDeliveryException {
+ Status result = Status.READY;
+ Channel channel = getChannel();
+ Transaction transaction = null;
+
+ try {
+ long processedEvents = 0;
+
+ transaction = channel.getTransaction();
+ transaction.begin();
+
+ // Preserve insertion order per collection but batch by target collection
+ // so that a single insertMany() call can be issued per collection.
+ Map<String, List<Document>> documentsByCollection = new LinkedHashMap<>();
+
+ for (; processedEvents < batchSize; processedEvents += 1) {
+ Event event = channel.take();
+
+ if (event == null) {
+ // no events available in channel
+ if (processedEvents == 0) {
+ result = Status.BACKOFF;
+ counter.incrementBatchEmptyCount();
+ } else {
+ counter.incrementBatchUnderflowCount();
+ }
+ break;
+ }
+ counter.incrementEventDrainAttemptCount();
+
+ String targetCollection = resolveCollection(event);
+ Document document = toDocument(event);
+
+ documentsByCollection
+ .computeIfAbsent(targetCollection, k -> new ArrayList<>())
+ .add(document);
+ }
+
+ for (Map.Entry<String, List<Document>> entry : documentsByCollection.entrySet()) {
+ writer.write(entry.getKey(), entry.getValue());
+ }
+
+ if (processedEvents > 0) {
+ counter.addToEventDrainSuccessCount(processedEvents);
+ }
+
+ transaction.commit();
+ } catch (Exception ex) {
+ String errorMsg = "Failed to publish events";
+ logger.error(errorMsg, ex);
+ counter.incrementEventWriteOrChannelFail(ex);
+ if (transaction != null) {
+ transaction.rollback();
+ }
+ throw new EventDeliveryException(errorMsg, ex);
+ } finally {
+ if (transaction != null) {
+ transaction.close();
+ }
+ }
+
+ return result;
+ }
+
+ /**
+ * Determines the collection an event should be written to. If a
+ * collection header is configured and the event has a value for that
+ * header, the value is looked up in the collection map to obtain the
+ * actual collection name. If there is no mapping for the value,
+ * {@link #collectionMapFallback} controls whether the default collection
+ * is used or an exception is thrown. If no collection header is
+ * configured, or the event has no value for it, the default collection
+ * is used.
+ */
+ private String resolveCollection(Event event) throws EventDeliveryException {
+ if (collectionHeader == null) {
+ return defaultCollection;
+ }
+ String headerValue = event.getHeaders().get(collectionHeader);
+ if (headerValue == null) {
+ return defaultCollection;
+ }
+ String mappedCollection = collectionMap.get(headerValue);
+ if (mappedCollection != null) {
+ return mappedCollection;
+ }
+ if (collectionMapFallback) {
+ logger.debug(
+ "No collection mapping found for header {}={}. Using default collection {}",
+ collectionHeader,
+ headerValue,
+ defaultCollection);
+ return defaultCollection;
+ }
+ throw new EventDeliveryException(
+ "No collection mapping found for header " + collectionHeader + "=" + headerValue);
+ }
+
+ private Document toDocument(Event event) {
+ Document document;
+ String body = new String(event.getBody(), StandardCharsets.UTF_8);
+ try {
+ document = Document.parse(body);
+ } catch (Exception ex) {
+ document = new Document(MongoDbSinkConstants.BODY_FIELD, body);
+ }
+ if (includeHeaders) {
+ document.append(MongoDbSinkConstants.HEADERS_FIELD, new Document(event.getHeaders()));
+ }
+ return document;
+ }
+
+ @Override
+ public synchronized void start() {
+ mongoClient = MongoClients.create(connectionUri);
+ MongoDatabase mongoDatabase = mongoClient.getDatabase(databaseName);
+ writer = new DefaultMongoDbWriter(mongoDatabase, writeConcern);
+ counter.start();
+ super.start();
+ }
+
+ @Override
+ public synchronized void stop() {
+ try {
+ if (writer != null) {
+ writer.close();
+ }
+ if (mongoClient != null) {
+ mongoClient.close();
+ }
+ } catch (MongoException ex) {
+ logger.warn("Error closing MongoDB client", ex);
+ }
+ counter.stop();
+ logger.info("MongoDb Sink {} stopped. Metrics: {}", getName(), counter);
+ super.stop();
+ }
+
+ @Override
+ public void configure(Context context) {
+ connectionUri = context.getString(MongoDbSinkConstants.CONNECTION_URI);
+ if (connectionUri == null || connectionUri.isEmpty()) {
+ throw new ConfigurationException("mongodb.uri must be specified");
+ }
+
+ databaseName = context.getString(MongoDbSinkConstants.DATABASE_NAME);
+ if (databaseName == null || databaseName.isEmpty()) {
+ throw new ConfigurationException("mongodb.database must be specified");
+ }
+
+ defaultCollection = context.getString(MongoDbSinkConstants.COLLECTION);
+ if (defaultCollection == null || defaultCollection.isEmpty()) {
+ throw new ConfigurationException("mongodb.collection must be specified");
+ }
+
+ collectionHeader = context.getString(MongoDbSinkConstants.COLLECTION_HEADER);
+ collectionMap = context.getSubProperties(MongoDbSinkConstants.COLLECTION_MAP_PREFIX);
+ collectionMapFallback = context.getBoolean(
+ MongoDbSinkConstants.COLLECTION_MAP_FALLBACK, MongoDbSinkConstants.DEFAULT_COLLECTION_MAP_FALLBACK);
+
+ if (collectionHeader != null && logger.isDebugEnabled()) {
+ logger.debug(
+ "Using header {} with mappings {} to select target collection", collectionHeader, collectionMap);
+ }
+
+ includeHeaders =
+ context.getBoolean(MongoDbSinkConstants.INCLUDE_HEADERS, MongoDbSinkConstants.DEFAULT_INCLUDE_HEADERS);
+
+ batchSize = context.getInteger(MongoDbSinkConstants.BATCH_SIZE, MongoDbSinkConstants.DEFAULT_BATCH_SIZE);
+
+ String writeConcernName = context.getString(MongoDbSinkConstants.WRITE_CONCERN);
+ if (writeConcernName != null && !writeConcernName.isEmpty()) {
+ writeConcern = WriteConcern.valueOf(writeConcernName);
+ if (writeConcern == null) {
+ throw new ConfigurationException("Unknown write concern: " + writeConcernName);
+ }
+ }
+
+ if (counter == null) {
+ counter = new SinkCounter(getName());
+ }
+ }
+}
diff --git a/flume-mongodb-sink/src/main/java/org/apache/flume/sink/mongodb/MongoDbSinkConstants.java b/flume-mongodb-sink/src/main/java/org/apache/flume/sink/mongodb/MongoDbSinkConstants.java
new file mode 100644
index 0000000..8ce0e97
--- /dev/null
+++ b/flume-mongodb-sink/src/main/java/org/apache/flume/sink/mongodb/MongoDbSinkConstants.java
@@ -0,0 +1,71 @@
+/*
+ * 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.flume.sink.mongodb;
+
+public class MongoDbSinkConstants {
+
+ public static final String MONGODB_PREFIX = "mongodb.";
+
+ /* Properties */
+
+ /** Mongo connection URI, e.g. mongodb://host1,host2/?replicaSet=rs0 */
+ public static final String CONNECTION_URI = MONGODB_PREFIX + "uri";
+
+ /** The database that documents will be written to. */
+ public static final String DATABASE_NAME = MONGODB_PREFIX + "database";
+
+ /** The default/fallback collection that documents will be written to. */
+ public static final String COLLECTION = MONGODB_PREFIX + "collection";
+
+ /**
+ * The name of the Flume event header whose value is used to select the
+ * target collection via {@link #COLLECTION_MAP_PREFIX}.
+ */
+ public static final String COLLECTION_HEADER = MONGODB_PREFIX + "collectionHeader";
+
+ /**
+ * Prefix for sub-properties that map a header value to a predefined
+ * collection name, e.g. mongodb.collectionMap.foo = fooCollection
+ */
+ public static final String COLLECTION_MAP_PREFIX = MONGODB_PREFIX + "collectionMap.";
+
+ /**
+ * Whether to fall back to the default collection when the header value
+ * does not have an entry in the collection map. If false, events with an
+ * unmapped header value cause the batch to fail.
+ */
+ public static final String COLLECTION_MAP_FALLBACK = MONGODB_PREFIX + "collectionMapFallback";
+
+ public static final boolean DEFAULT_COLLECTION_MAP_FALLBACK = true;
+
+ /**
+ * Whether the Flume event headers should be included in the resulting
+ * Mongo document, under the {@link #HEADERS_FIELD} field name.
+ */
+ public static final String INCLUDE_HEADERS = MONGODB_PREFIX + "includeHeaders";
+
+ public static final boolean DEFAULT_INCLUDE_HEADERS = false;
+
+ public static final String HEADERS_FIELD = "headers";
+ public static final String BODY_FIELD = "message";
+
+ public static final String BATCH_SIZE = "batchSize";
+ public static final int DEFAULT_BATCH_SIZE = 100;
+
+ /** Write concern to use, e.g. ACKNOWLEDGED, MAJORITY, JOURNALED, UNACKNOWLEDGED. */
+ public static final String WRITE_CONCERN = MONGODB_PREFIX + "writeConcern";
+}
diff --git a/flume-mongodb-sink/src/main/java/org/apache/flume/sink/mongodb/MongoDbWriter.java b/flume-mongodb-sink/src/main/java/org/apache/flume/sink/mongodb/MongoDbWriter.java
new file mode 100644
index 0000000..005fb53
--- /dev/null
+++ b/flume-mongodb-sink/src/main/java/org/apache/flume/sink/mongodb/MongoDbWriter.java
@@ -0,0 +1,43 @@
+/*
+ * 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.flume.sink.mongodb;
+
+import java.util.List;
+import org.bson.Document;
+
+/**
+ * Abstraction over the MongoDB write path used by {@link MongoDbSink}.
+ * Separating this out of the sink keeps the routing/document-construction
+ * logic in {@link MongoDbSink} independent of, and easily testable without,
+ * a live MongoDB connection.
+ */
+public interface MongoDbWriter {
+
+ /**
+ * Writes the given documents to the named collection.
+ *
+ * @param collectionName the target collection name
+ * @param documents the documents to insert, in order
+ */
+ void write(String collectionName, List<Document> documents);
+
+ /**
+ * Releases any resources (e.g. the underlying MongoDB client) held by
+ * this writer.
+ */
+ void close();
+}
diff --git a/flume-mongodb-sink/src/test/java/org/apache/flume/sink/mongodb/TestMongoDbSink.java b/flume-mongodb-sink/src/test/java/org/apache/flume/sink/mongodb/TestMongoDbSink.java
new file mode 100644
index 0000000..a2d2846
--- /dev/null
+++ b/flume-mongodb-sink/src/test/java/org/apache/flume/sink/mongodb/TestMongoDbSink.java
@@ -0,0 +1,261 @@
+/*
+ * 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.flume.sink.mongodb;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.flume.Channel;
+import org.apache.flume.Context;
+import org.apache.flume.EventDeliveryException;
+import org.apache.flume.Sink;
+import org.apache.flume.Transaction;
+import org.apache.flume.channel.MemoryChannel;
+import org.apache.flume.conf.Configurables;
+import org.apache.flume.conf.ConfigurationException;
+import org.apache.flume.event.EventBuilder;
+import org.apache.flume.instrumentation.SinkCounter;
+import org.bson.Document;
+import org.junit.Test;
+import org.mockito.internal.util.reflection.Whitebox;
+
+public class TestMongoDbSink {
+
+ /**
+ * In-memory {@link MongoDbWriter} fake used to capture what the sink
+ * would have written, without needing a live MongoDB connection.
+ */
+ private static final class FakeMongoDbWriter implements MongoDbWriter {
+ private final Map<String, List<Document>> written = new LinkedHashMap<>();
+ private boolean closed = false;
+
+ @Override
+ public void write(String collectionName, List<Document> documents) {
+ written.computeIfAbsent(collectionName, k -> new ArrayList<>()).addAll(documents);
+ }
+
+ @Override
+ public void close() {
+ closed = true;
+ }
+ }
+
+ private static Context baseContext() {
+ Context context = new Context();
+ context.put(MongoDbSinkConstants.CONNECTION_URI, "mongodb://localhost:27017");
+ context.put(MongoDbSinkConstants.DATABASE_NAME, "testDb");
+ context.put(MongoDbSinkConstants.COLLECTION, "defaultCollection");
+ return context;
+ }
+
+ private static MongoDbSink createSink(Context context, MongoDbWriter writer) {
+ MongoDbSink sink = new MongoDbSink();
+ Channel channel = new MemoryChannel();
+ Configurables.configure(channel, new Context());
+ sink.setChannel(channel);
+ channel.start();
+ Configurables.configure(sink, context);
+ Whitebox.setInternalState(sink, "writer", writer);
+ Whitebox.setInternalState(sink, "counter", new SinkCounter("test"));
+ return sink;
+ }
+
+ private static void putEvent(Channel channel, byte[] body, Map<String, String> headers) {
+ Transaction tx = channel.getTransaction();
+ tx.begin();
+ channel.put(EventBuilder.withBody(body, headers));
+ tx.commit();
+ tx.close();
+ }
+
+ @Test(expected = ConfigurationException.class)
+ public void testConfigureMissingUri() {
+ Context context = new Context();
+ context.put(MongoDbSinkConstants.DATABASE_NAME, "testDb");
+ context.put(MongoDbSinkConstants.COLLECTION, "col");
+ new MongoDbSink().configure(context);
+ }
+
+ @Test(expected = ConfigurationException.class)
+ public void testConfigureMissingDatabase() {
+ Context context = new Context();
+ context.put(MongoDbSinkConstants.CONNECTION_URI, "mongodb://localhost:27017");
+ context.put(MongoDbSinkConstants.COLLECTION, "col");
+ new MongoDbSink().configure(context);
+ }
+
+ @Test(expected = ConfigurationException.class)
+ public void testConfigureMissingCollection() {
+ Context context = new Context();
+ context.put(MongoDbSinkConstants.CONNECTION_URI, "mongodb://localhost:27017");
+ context.put(MongoDbSinkConstants.DATABASE_NAME, "testDb");
+ new MongoDbSink().configure(context);
+ }
+
+ @Test
+ public void testWritesToDefaultCollectionWhenNoHeaderConfigured() throws EventDeliveryException {
+ FakeMongoDbWriter writer = new FakeMongoDbWriter();
+ Context context = baseContext();
+ MongoDbSink sink = createSink(context, writer);
+ Channel channel = sink.getChannel();
+
+ putEvent(channel, "{\"foo\":\"bar\"}".getBytes(StandardCharsets.UTF_8), new HashMap<String, String>());
+
+ Sink.Status status = sink.process();
+ assertEquals(Sink.Status.READY, status);
+
+ List<Document> docs = writer.written.get("defaultCollection");
+ assertEquals(1, docs.size());
+ assertEquals("bar", docs.get(0).getString("foo"));
+ }
+
+ @Test
+ public void testHeaderMapsToPredefinedCollection() throws EventDeliveryException {
+ FakeMongoDbWriter writer = new FakeMongoDbWriter();
+ Context context = baseContext();
+ context.put(MongoDbSinkConstants.COLLECTION_HEADER, "type");
+ context.put(MongoDbSinkConstants.COLLECTION_MAP_PREFIX + "typeA", "collectionA");
+
+ MongoDbSink sink = createSink(context, writer);
+ Channel channel = sink.getChannel();
+
+ Map<String, String> headers = new HashMap<>();
+ headers.put("type", "typeA");
+ putEvent(channel, "{\"foo\":\"bar\"}".getBytes(StandardCharsets.UTF_8), headers);
+
+ Sink.Status status = sink.process();
+ assertEquals(Sink.Status.READY, status);
+
+ assertTrue(writer.written.containsKey("collectionA"));
+ assertFalse(writer.written.containsKey("defaultCollection"));
+ assertEquals(1, writer.written.get("collectionA").size());
+ }
+
+ @Test
+ public void testUnmappedHeaderFallsBackToDefaultCollection() throws EventDeliveryException {
+ FakeMongoDbWriter writer = new FakeMongoDbWriter();
+ Context context = baseContext();
+ context.put(MongoDbSinkConstants.COLLECTION_HEADER, "type");
+ context.put(MongoDbSinkConstants.COLLECTION_MAP_PREFIX + "typeA", "collectionA");
+
+ MongoDbSink sink = createSink(context, writer);
+ Channel channel = sink.getChannel();
+
+ Map<String, String> headers = new HashMap<>();
+ headers.put("type", "typeUnknown");
+ putEvent(channel, "{\"foo\":\"bar\"}".getBytes(StandardCharsets.UTF_8), headers);
+
+ Sink.Status status = sink.process();
+ assertEquals(Sink.Status.READY, status);
+
+ assertTrue(writer.written.containsKey("defaultCollection"));
+ assertFalse(writer.written.containsKey("collectionA"));
+ }
+
+ @Test
+ public void testUnmappedHeaderFailsWhenFallbackDisabled() throws EventDeliveryException {
+ FakeMongoDbWriter writer = new FakeMongoDbWriter();
+ Context context = baseContext();
+ context.put(MongoDbSinkConstants.COLLECTION_HEADER, "type");
+ context.put(MongoDbSinkConstants.COLLECTION_MAP_PREFIX + "typeA", "collectionA");
+ context.put(MongoDbSinkConstants.COLLECTION_MAP_FALLBACK, "false");
+
+ MongoDbSink sink = createSink(context, writer);
+ Channel channel = sink.getChannel();
+
+ Map<String, String> headers = new HashMap<>();
+ headers.put("type", "typeUnknown");
+ putEvent(channel, "{\"foo\":\"bar\"}".getBytes(StandardCharsets.UTF_8), headers);
+
+ try {
+ sink.process();
+ fail("Expected EventDeliveryException");
+ } catch (EventDeliveryException expected) {
+ assertTrue(expected.getCause().getMessage().contains("No collection mapping"));
+ }
+ assertTrue(writer.written.isEmpty());
+ }
+
+ @Test
+ public void testNonJsonBodyIsWrappedInMessageField() throws EventDeliveryException {
+ FakeMongoDbWriter writer = new FakeMongoDbWriter();
+ Context context = baseContext();
+ MongoDbSink sink = createSink(context, writer);
+ Channel channel = sink.getChannel();
+
+ putEvent(channel, "plain text event".getBytes(StandardCharsets.UTF_8), new HashMap<String, String>());
+
+ sink.process();
+
+ List<Document> docs = writer.written.get("defaultCollection");
+ assertEquals("plain text event", docs.get(0).getString(MongoDbSinkConstants.BODY_FIELD));
+ }
+
+ @Test
+ public void testIncludeHeadersAddsHeadersField() throws EventDeliveryException {
+ FakeMongoDbWriter writer = new FakeMongoDbWriter();
+ Context context = baseContext();
+ context.put(MongoDbSinkConstants.INCLUDE_HEADERS, "true");
+ MongoDbSink sink = createSink(context, writer);
+ Channel channel = sink.getChannel();
+
+ Map<String, String> headers = new HashMap<>();
+ headers.put("source", "app1");
+ putEvent(channel, "{\"foo\":\"bar\"}".getBytes(StandardCharsets.UTF_8), headers);
+
+ sink.process();
+
+ Document doc = writer.written.get("defaultCollection").get(0);
+ Document headerDoc = (Document) doc.get(MongoDbSinkConstants.HEADERS_FIELD);
+ assertEquals("app1", headerDoc.getString("source"));
+ }
+
+ @Test
+ public void testBackoffOnEmptyChannel() throws EventDeliveryException {
+ FakeMongoDbWriter writer = new FakeMongoDbWriter();
+ Context context = baseContext();
+ MongoDbSink sink = createSink(context, writer);
+
+ Sink.Status status = sink.process();
+ assertEquals(Sink.Status.BACKOFF, status);
+ assertTrue(writer.written.isEmpty());
+ }
+
+ @Test
+ public void testCollectionMapParsedFromSubProperties() {
+ Context context = baseContext();
+ context.put(MongoDbSinkConstants.COLLECTION_HEADER, "type");
+ context.put(MongoDbSinkConstants.COLLECTION_MAP_PREFIX + "typeA", "collectionA");
+ context.put(MongoDbSinkConstants.COLLECTION_MAP_PREFIX + "typeB", "collectionB");
+
+ MongoDbSink sink = new MongoDbSink();
+ sink.configure(context);
+
+ @SuppressWarnings("unchecked")
+ Map<String, String> collectionMap = (Map<String, String>) Whitebox.getInternalState(sink, "collectionMap");
+ assertEquals("collectionA", collectionMap.get("typeA"));
+ assertEquals("collectionB", collectionMap.get("typeB"));
+ }
+}
diff --git a/flume-mongodb-sink/src/test/resources/log4j2.xml b/flume-mongodb-sink/src/test/resources/log4j2.xml
new file mode 100644
index 0000000..6d5c01b
--- /dev/null
+++ b/flume-mongodb-sink/src/test/resources/log4j2.xml
@@ -0,0 +1,71 @@
+<?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.
+ -->
+<Configuration status="OFF">
+ <Properties>
+ <Property name="kafka.logs.dir">target/logs</Property>
+ </Properties>
+ <Appenders>
+ <Console name="Console" target="SYSTEM_OUT">
+ <PatternLayout pattern="%d (%t) [%p - %l] %m%n" />
+ </Console>
+ <RollingFile name="kafka" fileName="${kafka.logs.dir}/server.log" filePattern="${kafka.logs.dir}/server.log.%i">
+ <PatternLayout pattern="[%d] %p %m (%c)%n" />
+ <CronTriggeringPolicy schedule="0 0 0 * * ?"/>
+ </RollingFile>
+ <RollingFile name="stateChange" fileName="${kafka.logs.dir}/state-change.log" filePattern="${kafka.logs.dir}/state-change.log.%i">
+ <PatternLayout pattern="[%d] %p %m (%c)%n" />
+ <CronTriggeringPolicy schedule="0 0 0 * * ?"/>
+ </RollingFile>
+ <RollingFile name="request" fileName="${kafka.logs.dir}/kafka-request.log" filePattern="${kafka.logs.dir}/kafka-request.log.%i">
+ <PatternLayout pattern="[%d] %p %m (%c)%n" />
+ <CronTriggeringPolicy schedule="0 0 0 * * ?"/>
+ </RollingFile>
+ <RollingFile name="cleaner" fileName="${kafka.logs.dir}/log-cleaner.log" filePattern="${kafka.logs.dir}/log-cleaner.log.%i">
+ <PatternLayout pattern="[%d] %p %m (%c)%n" />
+ <CronTriggeringPolicy schedule="0 0 0 * * ?"/>
+ </RollingFile>
+ <RollingFile name="controller" fileName="${kafka.logs.dir}/controller.log" filePattern="${kafka.logs.dir}/controller.log.%i">
+ <PatternLayout pattern="[%d] %p %m (%c)%n" />
+ <CronTriggeringPolicy schedule="0 0 0 * * ?"/>
+ </RollingFile>
+ </Appenders>
+
+ <Loggers>
+ <Logger name="kafka" level="info">
+ <AppenderRef ref="kafka"/>
+ </Logger>
+ <Logger name="kafka.network.RequestChannel.*" level="WARN" additivity="false">
+ <AppenderRef ref="request"/>
+ </Logger>
+ <Logger name="kafka.request.logger" level="WARN" additivity="false">
+ <AppenderRef ref="request"/>
+ </Logger>
+ <Logger name="kafka.controller" level="TRACE" additivity="false">
+ <AppenderRef ref="controller"/>
+ </Logger>
+ <Logger name="kafka.log.cleaner" level="INFO" additivity="false">
+ <AppenderRef ref="cleaner"/>
+ </Logger>
+ <Logger name="state.change.logger" level="TRACE" additivity="false">
+ <AppenderRef ref="stateChange"/>
+ </Logger>
+ <Root level="INFO">
+ <AppenderRef ref="Console" />
+ </Root>
+ </Loggers>
+</Configuration>
diff --git a/pom.xml b/pom.xml
new file mode 100644
index 0000000..9567c93
--- /dev/null
+++ b/pom.xml
@@ -0,0 +1,229 @@
+<?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 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.flume</groupId>
+ <artifactId>flume-parent</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ <relativePath />
+ </parent>
+
+ <groupId>org.apache.flume</groupId>
+ <artifactId>flume-mongodb-parent</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ <packaging>pom</packaging>
+ <name>Flume MongoDB Parent</name>
+
+ <inceptionYear>2022</inceptionYear>
+
+ <organization>
+ <name>Apache Software Foundation</name>
+ <url>http://www.apache.org</url>
+ </organization>
+
+ <licenses>
+ <license>
+ <name>The Apache Software License, Version 2.0</name>
+ <url>http://www.apache.org/licenses/LICENSE-2.0.txt</url>
+ </license>
+ </licenses>
+
+ <developers>
+ <developer>
+ <id>rgoers</id>
+ <name>Ralph Goers</name>
+ <email>rgoers@apache.org</email>
+ <organization>Intuit</organization>
+ </developer>
+ </developers>
+
+ <mailingLists>
+ <mailingList>
+ <name>Flume User List</name>
+ <subscribe>user-subscribe@flume.apache.org</subscribe>
+ <unsubscribe>user-unsubscribe@flume.apache.org</unsubscribe>
+ <post>user@flume.apache.org</post>
+ <archive>http://mail-archives.apache.org/mod_mbox/flume-user/</archive>
+ </mailingList>
+ <mailingList>
+ <name>Flume Developer List</name>
+ <subscribe>dev-subscribe@flume.apache.org</subscribe>
+ <unsubscribe>dev-unsubscribe@flume.apache.org</unsubscribe>
+ <post>dev@flume.apache.org</post>
+ <archive>http://mail-archives.apache.org/mod_mbox/flume-dev/</archive>
+ </mailingList>
+ <mailingList>
+ <name>Flume Commits</name>
+ <subscribe>commits-subscribe@flume.apache.org</subscribe>
+ <unsubscribe>commits-unsubscribe@flume.apache.org</unsubscribe>
+ <post>commits@flume.apache.org</post>
+ <archive>http://mail-archives.apache.org/mod_mbox/flume-commits/</archive>
+ </mailingList>
+ </mailingLists>
+ <modules>
+ <module>flume-mongodb-sink</module>
+ </modules>
+
+ <scm>
+ <connection>https://gitbox.apache.org/repos/asf/flume-spring-boot.git</connection>
+ <developerConnection>https://gitbox.apache.org/repos/asf/flume-spring-boot.git</developerConnection>
+ <url>https://gitbox.apache.org/repos/asf/flume-spring-boot.git</url>
+ </scm>
+
+ <issueManagement>
+ <system>JIRA</system>
+ <url>https://issues.apache.org/jira/browse/FLUME</url>
+ </issueManagement>
+
+ <properties>
+ <ReleaseVersion>2.0.0</ReleaseVersion>
+ <ReleaseManager>Ralph Goers</ReleaseManager>
+ <ReleaseKey>B3D8E1BA</ReleaseKey>
+ <SigningUserName>rgoers@apache.org</SigningUserName>
+ <checksum-maven-plugin.version>1.11</checksum-maven-plugin.version>
+ <dropwizard-metrics.version>4.1.18</dropwizard-metrics.version>
+ <findsecbugs-plugin.version>1.12.0</findsecbugs-plugin.version>
+ <flume.version>2.0.0-SNAPSHOT</flume.version>
+ <jackson.version>2.17.0</jackson.version>
+ <junit.version>4.13.2</junit.version>
+ <mongodb.version>3.7.0</mongodb.version>
+ <log4j.version>2.26.1</log4j.version>
+ <maven.compiler.source>11</maven.compiler.source>
+ <maven.compiler.target>11</maven.compiler.target>
+ <mockito.version>1.9.0</mockito.version>
+ <module.name>org.apache.flume.mongodb</module.name>
+ <mvn-gpg-plugin.version>1.6</mvn-gpg-plugin.version>
+ <mvn-javadoc-plugin.version>2.9</mvn-javadoc-plugin.version>
+ <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+ <rat.version>0.12</rat.version>
+ <spotbugs-maven-plugin.version>4.7.2.1</spotbugs-maven-plugin.version>
+ <spotless-maven-plugin.version>2.27.2</spotless-maven-plugin.version>
+ <mongodb.version>5.10.0</mongodb.version>
+ </properties>
+ <dependencyManagement>
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.flume</groupId>
+ <artifactId>flume-ng-core</artifactId>
+ <version>${flume.version}</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-configuration</artifactId>
+ <version>${flume.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.mongodb</groupId>
+ <artifactId>mongodb-driver-bom</artifactId>
+ <version>${mongodb.version}</version>
+ <type>pom</type>
+ <scope>import</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>io.dropwizard.metrics</groupId>
+ <artifactId>metrics-core</artifactId>
+ <version>${dropwizard-metrics.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-api</artifactId>
+ <version>${log4j.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-core</artifactId>
+ <version>${log4j.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-core</artifactId>
+ <version>${jackson.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <version>${junit.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-all</artifactId>
+ <version>${mockito.version}</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+ </dependencyManagement>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.rat</groupId>
+ <artifactId>apache-rat-plugin</artifactId>
+ <version>${rat.version}</version>
+ <configuration>
+ <excludes>
+ <exclude>**/.idea/</exclude>
+ <exclude>**/*.iml</exclude>
+ <exclude>src/main/resources/META-INF/services/**/*</exclude>
+ <exclude>**/nb-configuration.xml</exclude>
+ <exclude>.git/</exclude>
+ <exclude>patchprocess/</exclude>
+ <exclude>.gitignore</exclude>
+ <exclude>**/*.yml</exclude>
+ <exclude>**/*.yaml</exclude>
+ <exclude>**/*.json</exclude>
+ <!-- ASF jenkins box puts the Maven repo in our root directory. -->
+ <exclude>.repository/</exclude>
+ <exclude>**/*.diff</exclude>
+ <exclude>**/*.patch</exclude>
+ <exclude>**/*.avsc</exclude>
+ <exclude>**/*.avro</exclude>
+ <exclude>**/docs/**</exclude>
+ <exclude>**/test/resources/**</exclude>
+ <exclude>**/.settings/*</exclude>
+ <exclude>**/.classpath</exclude>
+ <exclude>**/.project</exclude>
+ <exclude>**/target/**</exclude>
+ <exclude>**/derby.log</exclude>
+ <exclude>**/metastore_db/</exclude>
+ <exclude>.mvn/**</exclude>
+ <exclude>**/exclude-pmd.properties</exclude>
+ </excludes>
+ <consoleOutput>true</consoleOutput>
+ </configuration>
+ <executions>
+ <execution>
+ <id>verify.rat</id>
+ <goals>
+ <goal>check</goal>
+ </goals>
+ <phase>verify</phase>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+</project>