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 <[email protected]> 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: [email protected] + 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>[email protected]</email> + <organization>Intuit</organization> + </developer> + </developers> + + <mailingLists> + <mailingList> + <name>Flume User List</name> + <subscribe>[email protected]</subscribe> + <unsubscribe>[email protected]</unsubscribe> + <post>[email protected]</post> + <archive>http://mail-archives.apache.org/mod_mbox/flume-user/</archive> + </mailingList> + <mailingList> + <name>Flume Developer List</name> + <subscribe>[email protected]</subscribe> + <unsubscribe>[email protected]</unsubscribe> + <post>[email protected]</post> + <archive>http://mail-archives.apache.org/mod_mbox/flume-dev/</archive> + </mailingList> + <mailingList> + <name>Flume Commits</name> + <subscribe>[email protected]</subscribe> + <unsubscribe>[email protected]</unsubscribe> + <post>[email protected]</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>[email protected]</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>
