This is an automated email from the ASF dual-hosted git repository.
samaitra pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/master by this push:
new 3ac680f IGNITE-12361 Migrate Flume module to ignite-extensions -
Fixes #7227.
3ac680f is described below
commit 3ac680f5bae0c48a4740f935d07ace9e2a4d5049
Author: samaitra <[email protected]>
AuthorDate: Sun Apr 12 14:01:21 2020 -0500
IGNITE-12361 Migrate Flume module to ignite-extensions - Fixes #7227.
Signed-off-by: samaitra <[email protected]>
---
modules/flume/README.txt | 48 -----
modules/flume/licenses/apache-2.0.txt | 202 ---------------------
modules/flume/pom.xml | 88 ---------
.../ignite/stream/flume/EventTransformer.java | 36 ----
.../org/apache/ignite/stream/flume/IgniteSink.java | 196 --------------------
.../ignite/stream/flume/IgniteSinkConstants.java | 35 ----
.../apache/ignite/stream/flume/package-info.java | 22 ---
.../apache/ignite/stream/flume/IgniteSinkTest.java | 139 --------------
.../ignite/stream/flume/IgniteSinkTestSuite.java | 29 ---
.../ignite/stream/flume/TestEventTransformer.java | 66 -------
.../apache/ignite/stream/flume/package-info.java | 22 ---
.../flume/src/test/resources/example-ignite.xml | 71 --------
parent/pom.xml | 1 -
pom.xml | 1 -
14 files changed, 956 deletions(-)
diff --git a/modules/flume/README.txt b/modules/flume/README.txt
deleted file mode 100644
index adcd021..0000000
--- a/modules/flume/README.txt
+++ /dev/null
@@ -1,48 +0,0 @@
-Apache Ignite Flume Sink Module
--------------------------------
-
-IgniteSink is a Flume sink that extracts Events from an associated Flume
channel and injects into an Ignite cache.
-Flume 1.7.0 is supported.
-
-IgniteSink, which can be found in 'optional/ignite-flume', and its
dependencies have to be included in the agent's classpath,
-as described in the following subsection, before starting the Flume agent.
-
-## Setting up and running
-
-1. Create a transformer by implementing EventTransformer interface.
-2. Create 'ignite' directory inside plugins.d directory which is located in
${FLUME_HOME}. If the plugins.d directory is not there, create it.
-3. Build it and copy to ${FLUME_HOME}/plugins.d/ignite-sink/lib.
-4. Copy other Ignite-related jar files from Apache Ignite distribution to
${FLUME_HOME}/plugins.d/ignite-sink/libext to have them as shown below.
-
-```
-plugins.d/
-`-- ignite
- |-- lib
- | `-- ignite-flume-transformer-x.x.x.jar <-- your jar
- `-- libext
- |-- cache-api-1.0.0.jar
- |-- ignite-core-x.x.x.jar
- |-- ignite-flume-x.x.x.jar <-- IgniteSink
- |-- ignite-spring-x.x.x.jar
- |-- spring-aop-4.1.0.RELEASE.jar
- |-- spring-beans-4.1.0.RELEASE.jar
- |-- spring-context-4.1.0.RELEASE.jar
- |-- spring-core-4.1.0.RELEASE.jar
- `-- spring-expression-4.1.0.RELEASE.jar
-```
-
-4. In Flume configuration file, specify Ignite configuration XML file's
location with cache properties
-(see [Apache Ignite](https://apacheignite.readme.io/) with cache name
specified for cache creation,
-cache name (same as in Ignite configuration file), your EventTransformer's
implementation class,
-and, optionally, batch size (default -- 100).
-
-```
-# Describe the sink
-a1.sinks.k1.type = org.apache.ignite.stream.flume.IgniteSink
-a1.sinks.k1.igniteCfg = /some-path/ignite.xml
-a1.sinks.k1.cacheName = testCache
-a1.sinks.k1.eventTransformer = my.company.MyEventTransformer
-a1.sinks.k1.batchSize = 100
-```
-
-After specifying your source and channel (see Flume's docs), you are ready to
run a Flume agent.
diff --git a/modules/flume/licenses/apache-2.0.txt
b/modules/flume/licenses/apache-2.0.txt
deleted file mode 100644
index d645695..0000000
--- a/modules/flume/licenses/apache-2.0.txt
+++ /dev/null
@@ -1,202 +0,0 @@
-
- Apache License
- Version 2.0, January 2004
- http://www.apache.org/licenses/
-
- TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
-
- 1. Definitions.
-
- "License" shall mean the terms and conditions for use, reproduction,
- and distribution as defined by Sections 1 through 9 of this document.
-
- "Licensor" shall mean the copyright owner or entity authorized by
- the copyright owner that is granting the License.
-
- "Legal Entity" shall mean the union of the acting entity and all
- other entities that control, are controlled by, or are under common
- control with that entity. For the purposes of this definition,
- "control" means (i) the power, direct or indirect, to cause the
- direction or management of such entity, whether by contract or
- otherwise, or (ii) ownership of fifty percent (50%) or more of the
- outstanding shares, or (iii) beneficial ownership of such entity.
-
- "You" (or "Your") shall mean an individual or Legal Entity
- exercising permissions granted by this License.
-
- "Source" form shall mean the preferred form for making modifications,
- including but not limited to software source code, documentation
- source, and configuration files.
-
- "Object" form shall mean any form resulting from mechanical
- transformation or translation of a Source form, including but
- not limited to compiled object code, generated documentation,
- and conversions to other media types.
-
- "Work" shall mean the work of authorship, whether in Source or
- Object form, made available under the License, as indicated by a
- copyright notice that is included in or attached to the work
- (an example is provided in the Appendix below).
-
- "Derivative Works" shall mean any work, whether in Source or Object
- form, that is based on (or derived from) the Work and for which the
- editorial revisions, annotations, elaborations, or other modifications
- represent, as a whole, an original work of authorship. For the purposes
- of this License, Derivative Works shall not include works that remain
- separable from, or merely link (or bind by name) to the interfaces of,
- the Work and Derivative Works thereof.
-
- "Contribution" shall mean any work of authorship, including
- the original version of the Work and any modifications or additions
- to that Work or Derivative Works thereof, that is intentionally
- submitted to Licensor for inclusion in the Work by the copyright owner
- or by an individual or Legal Entity authorized to submit on behalf of
- the copyright owner. For the purposes of this definition, "submitted"
- means any form of electronic, verbal, or written communication sent
- to the Licensor or its representatives, including but not limited to
- communication on electronic mailing lists, source code control systems,
- and issue tracking systems that are managed by, or on behalf of, the
- Licensor for the purpose of discussing and improving the Work, but
- excluding communication that is conspicuously marked or otherwise
- designated in writing by the copyright owner as "Not a Contribution."
-
- "Contributor" shall mean Licensor and any individual or Legal Entity
- on behalf of whom a Contribution has been received by Licensor and
- subsequently incorporated within the Work.
-
- 2. Grant of Copyright License. Subject to the terms and conditions of
- this License, each Contributor hereby grants to You a perpetual,
- worldwide, non-exclusive, no-charge, royalty-free, irrevocable
- copyright license to reproduce, prepare Derivative Works of,
- publicly display, publicly perform, sublicense, and distribute the
- Work and such Derivative Works in Source or Object form.
-
- 3. Grant of Patent License. Subject to the terms and conditions of
- this License, each Contributor hereby grants to You a perpetual,
- worldwide, non-exclusive, no-charge, royalty-free, irrevocable
- (except as stated in this section) patent license to make, have made,
- use, offer to sell, sell, import, and otherwise transfer the Work,
- where such license applies only to those patent claims licensable
- by such Contributor that are necessarily infringed by their
- Contribution(s) alone or by combination of their Contribution(s)
- with the Work to which such Contribution(s) was submitted. If You
- institute patent litigation against any entity (including a
- cross-claim or counterclaim in a lawsuit) alleging that the Work
- or a Contribution incorporated within the Work constitutes direct
- or contributory patent infringement, then any patent licenses
- granted to You under this License for that Work shall terminate
- as of the date such litigation is filed.
-
- 4. Redistribution. You may reproduce and distribute copies of the
- Work or Derivative Works thereof in any medium, with or without
- modifications, and in Source or Object form, provided that You
- meet the following conditions:
-
- (a) You must give any other recipients of the Work or
- Derivative Works a copy of this License; and
-
- (b) You must cause any modified files to carry prominent notices
- stating that You changed the files; and
-
- (c) You must retain, in the Source form of any Derivative Works
- that You distribute, all copyright, patent, trademark, and
- attribution notices from the Source form of the Work,
- excluding those notices that do not pertain to any part of
- the Derivative Works; and
-
- (d) If the Work includes a "NOTICE" text file as part of its
- distribution, then any Derivative Works that You distribute must
- include a readable copy of the attribution notices contained
- within such NOTICE file, excluding those notices that do not
- pertain to any part of the Derivative Works, in at least one
- of the following places: within a NOTICE text file distributed
- as part of the Derivative Works; within the Source form or
- documentation, if provided along with the Derivative Works; or,
- within a display generated by the Derivative Works, if and
- wherever such third-party notices normally appear. The contents
- of the NOTICE file are for informational purposes only and
- do not modify the License. You may add Your own attribution
- notices within Derivative Works that You distribute, alongside
- or as an addendum to the NOTICE text from the Work, provided
- that such additional attribution notices cannot be construed
- as modifying the License.
-
- You may add Your own copyright statement to Your modifications and
- may provide additional or different license terms and conditions
- for use, reproduction, or distribution of Your modifications, or
- for any such Derivative Works as a whole, provided Your use,
- reproduction, and distribution of the Work otherwise complies with
- the conditions stated in this License.
-
- 5. Submission of Contributions. Unless You explicitly state otherwise,
- any Contribution intentionally submitted for inclusion in the Work
- by You to the Licensor shall be under the terms and conditions of
- this License, without any additional terms or conditions.
- Notwithstanding the above, nothing herein shall supersede or modify
- the terms of any separate license agreement you may have executed
- with Licensor regarding such Contributions.
-
- 6. Trademarks. This License does not grant permission to use the trade
- names, trademarks, service marks, or product names of the Licensor,
- except as required for reasonable and customary use in describing the
- origin of the Work and reproducing the content of the NOTICE file.
-
- 7. Disclaimer of Warranty. Unless required by applicable law or
- agreed to in writing, Licensor provides the Work (and each
- Contributor provides its Contributions) on an "AS IS" BASIS,
- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
- implied, including, without limitation, any warranties or conditions
- of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
- PARTICULAR PURPOSE. You are solely responsible for determining the
- appropriateness of using or redistributing the Work and assume any
- risks associated with Your exercise of permissions under this License.
-
- 8. Limitation of Liability. In no event and under no legal theory,
- whether in tort (including negligence), contract, or otherwise,
- unless required by applicable law (such as deliberate and grossly
- negligent acts) or agreed to in writing, shall any Contributor be
- liable to You for damages, including any direct, indirect, special,
- incidental, or consequential damages of any character arising as a
- result of this License or out of the use or inability to use the
- Work (including but not limited to damages for loss of goodwill,
- work stoppage, computer failure or malfunction, or any and all
- other commercial damages or losses), even if such Contributor
- has been advised of the possibility of such damages.
-
- 9. Accepting Warranty or Additional Liability. While redistributing
- the Work or Derivative Works thereof, You may choose to offer,
- and charge a fee for, acceptance of support, warranty, indemnity,
- or other liability obligations and/or rights consistent with this
- License. However, in accepting such obligations, You may act only
- on Your own behalf and on Your sole responsibility, not on behalf
- of any other Contributor, and only if You agree to indemnify,
- defend, and hold each Contributor harmless for any liability
- incurred by, or claims asserted against, such Contributor by reason
- of your accepting any such warranty or additional liability.
-
- END OF TERMS AND CONDITIONS
-
- APPENDIX: How to apply the Apache License to your work.
-
- To apply the Apache License to your work, attach the following
- boilerplate notice, with the fields enclosed by brackets "[]"
- replaced with your own identifying information. (Don't include
- the brackets!) The text should be enclosed in the appropriate
- comment syntax for the file format. We also recommend that a
- file or class name and description of purpose be included on the
- same "printed page" as the copyright notice for easier
- identification within third-party archives.
-
- Copyright [yyyy] [name of copyright owner]
-
- Licensed under the Apache License, Version 2.0 (the "License");
- you may not use this file except in compliance with the License.
- You may obtain a copy of the License at
-
- http://www.apache.org/licenses/LICENSE-2.0
-
- Unless required by applicable law or agreed to in writing, software
- distributed under the License is distributed on an "AS IS" BASIS,
- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- See the License for the specific language governing permissions and
- limitations under the License.
diff --git a/modules/flume/pom.xml b/modules/flume/pom.xml
deleted file mode 100644
index a86dbb1..0000000
--- a/modules/flume/pom.xml
+++ /dev/null
@@ -1,88 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-
-<!--
- Licensed to the Apache Software Foundation (ASF) under one or more
- contributor license agreements. See the NOTICE file distributed with
- this work for additional information regarding copyright ownership.
- The ASF licenses this file to You under the Apache License, Version 2.0
- (the "License"); you may not use this file except in compliance with
- the License. You may obtain a copy of the License at
-
- http://www.apache.org/licenses/LICENSE-2.0
-
- Unless required by applicable law or agreed to in writing, software
- distributed under the License is distributed on an "AS IS" BASIS,
- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- See the License for the specific language governing permissions and
- limitations under the License.
--->
-
-<!--
- POM file.
--->
-<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.ignite</groupId>
- <artifactId>ignite-parent</artifactId>
- <version>1</version>
- <relativePath>../../parent</relativePath>
- </parent>
-
- <artifactId>ignite-flume</artifactId>
- <version>2.9.0-SNAPSHOT</version>
- <url>http://ignite.apache.org</url>
-
- <dependencies>
- <dependency>
- <groupId>org.apache.ignite</groupId>
- <artifactId>ignite-core</artifactId>
- <version>${project.version}</version>
- </dependency>
-
- <dependency>
- <groupId>org.apache.flume</groupId>
- <artifactId>flume-ng-core</artifactId>
- <version>${flume.ng.version}</version>
- <exclusions>
- <exclusion>
- <groupId>commons-codec</groupId>
- <artifactId>commons-codec</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
-
- <dependency>
- <groupId>org.apache.ignite</groupId>
- <artifactId>ignite-spring</artifactId>
- <version>${project.version}</version>
- <scope>test</scope>
- </dependency>
-
- <dependency>
- <groupId>org.apache.ignite</groupId>
- <artifactId>ignite-log4j</artifactId>
- <version>${project.version}</version>
- <scope>test</scope>
- </dependency>
-
- <dependency>
- <groupId>org.apache.ignite</groupId>
- <artifactId>ignite-core</artifactId>
- <version>${project.version}</version>
- <type>test-jar</type>
- <scope>test</scope>
- </dependency>
- </dependencies>
-
- <build>
- <plugins>
- <!-- Generate the OSGi MANIFEST.MF for this bundle. -->
- <plugin>
- <groupId>org.apache.felix</groupId>
- <artifactId>maven-bundle-plugin</artifactId>
- </plugin>
- </plugins>
- </build>
-</project>
diff --git
a/modules/flume/src/main/java/org/apache/ignite/stream/flume/EventTransformer.java
b/modules/flume/src/main/java/org/apache/ignite/stream/flume/EventTransformer.java
deleted file mode 100644
index 913ebb4..0000000
---
a/modules/flume/src/main/java/org/apache/ignite/stream/flume/EventTransformer.java
+++ /dev/null
@@ -1,36 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.stream.flume;
-
-import java.util.List;
-import java.util.Map;
-import org.jetbrains.annotations.Nullable;
-
-/**
- * Flume event transformer to convert a list of Flume Events to cache entries.
- */
-public interface EventTransformer<Event, K, V> {
-
- /**
- * Transforms a list of Flume Events to cache entries.
- *
- * @param events List of Flume events to transform.
- * @return Cache entries to be written into the grid.
- */
- @Nullable Map<K, V> transform(List<Event> events);
-}
diff --git
a/modules/flume/src/main/java/org/apache/ignite/stream/flume/IgniteSink.java
b/modules/flume/src/main/java/org/apache/ignite/stream/flume/IgniteSink.java
deleted file mode 100644
index 54482e5..0000000
--- a/modules/flume/src/main/java/org/apache/ignite/stream/flume/IgniteSink.java
+++ /dev/null
@@ -1,196 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.stream.flume;
-
-import java.util.ArrayList;
-import java.util.List;
-import org.apache.flume.Channel;
-import org.apache.flume.Context;
-import org.apache.flume.Event;
-import org.apache.flume.EventDeliveryException;
-import org.apache.flume.FlumeException;
-import org.apache.flume.Transaction;
-import org.apache.flume.conf.Configurable;
-import org.apache.flume.instrumentation.SinkCounter;
-import org.apache.flume.sink.AbstractSink;
-import org.apache.ignite.Ignite;
-import org.apache.ignite.Ignition;
-import org.apache.ignite.internal.util.typedef.internal.A;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Flume sink for Apache Ignite.
- */
-public class IgniteSink extends AbstractSink implements Configurable {
- /** Logger. */
- private static final Logger log =
LoggerFactory.getLogger(IgniteSink.class);
-
- /** Default batch size. */
- private static final int DFLT_BATCH_SIZE = 100;
-
- /** Ignite configuration file. */
- private String springCfgPath;
-
- /** Cache name. */
- private String cacheName;
-
- /** Event transformer implementation class. */
- private String eventTransformerCls;
-
- /** Number of events to be written per Flume transaction. */
- private int batchSize;
-
- /** Monitoring counter. */
- private SinkCounter sinkCounter;
-
- /** Event transformer. */
- private EventTransformer<Event, Object, Object> eventTransformer;
-
- /** Ignite instance. */
- private Ignite ignite;
-
- /** Empty constructor. */
- public IgniteSink() {
- }
-
- /**
- * Sink configurations with Ignite-specific settings.
- *
- * @param context Context for sink.
- */
- @Override public void configure(Context context) {
- springCfgPath = context.getString(IgniteSinkConstants.CFG_PATH);
- cacheName = context.getString(IgniteSinkConstants.CFG_CACHE_NAME);
- eventTransformerCls =
context.getString(IgniteSinkConstants.CFG_EVENT_TRANSFORMER);
- batchSize = context.getInteger(IgniteSinkConstants.CFG_BATCH_SIZE,
DFLT_BATCH_SIZE);
-
- if (sinkCounter == null)
- sinkCounter = new SinkCounter(getName());
- }
-
- /**
- * Starts a grid and initializes an event transformer.
- */
- @SuppressWarnings("unchecked")
- @Override public synchronized void start() {
- A.notNull(springCfgPath, "Ignite config file");
- A.notNull(cacheName, "Cache name");
- A.notNull(eventTransformerCls, "Event transformer class");
-
- sinkCounter.start();
-
- try {
- if (ignite == null)
- ignite = Ignition.start(springCfgPath);
-
- if (eventTransformerCls != null && !eventTransformerCls.isEmpty())
{
- Class<? extends EventTransformer> clazz =
- (Class<? extends EventTransformer<Event, Object,
Object>>)Class.forName(eventTransformerCls);
-
- eventTransformer = clazz.newInstance();
- }
- }
- catch (Exception e) {
- log.error("Failed to start grid", e);
-
- sinkCounter.incrementConnectionFailedCount();
-
- throw new FlumeException("Failed to start grid", e);
- }
-
- sinkCounter.incrementConnectionCreatedCount();
-
- super.start();
- }
-
- /**
- * Stops the grid.
- */
- @Override public synchronized void stop() {
- if (ignite != null)
- ignite.close();
-
- sinkCounter.incrementConnectionClosedCount();
- sinkCounter.stop();
-
- super.stop();
- }
-
- /**
- * Processes Flume events.
- */
- @Override public Status process() throws EventDeliveryException {
- Channel channel = getChannel();
-
- Transaction transaction = channel.getTransaction();
-
- int eventCount = 0;
-
- try {
- transaction.begin();
-
- List<Event> batch = new ArrayList<>(batchSize);
-
- for (; eventCount < batchSize; ++eventCount) {
- Event event = channel.take();
-
- if (event == null) {
- break;
- }
-
- batch.add(event);
- }
-
- if (!batch.isEmpty()) {
-
ignite.cache(cacheName).putAll(eventTransformer.transform(batch));
-
- if (batch.size() < batchSize)
- sinkCounter.incrementBatchUnderflowCount();
- else
- sinkCounter.incrementBatchCompleteCount();
- }
- else {
- sinkCounter.incrementBatchEmptyCount();
- }
-
- sinkCounter.addToEventDrainAttemptCount(batch.size());
-
- transaction.commit();
-
- sinkCounter.addToEventDrainSuccessCount(batch.size());
- }
- catch (Exception e) {
- log.error("Failed to process events", e);
-
- try {
- transaction.rollback();
- }
- catch (Throwable e1) {
- e.addSuppressed(e1);
- }
-
- throw new EventDeliveryException(e);
- }
- finally {
- transaction.close();
- }
-
- return eventCount == 0 ? Status.BACKOFF : Status.READY;
- }
-}
diff --git
a/modules/flume/src/main/java/org/apache/ignite/stream/flume/IgniteSinkConstants.java
b/modules/flume/src/main/java/org/apache/ignite/stream/flume/IgniteSinkConstants.java
deleted file mode 100644
index ddefb24..0000000
---
a/modules/flume/src/main/java/org/apache/ignite/stream/flume/IgniteSinkConstants.java
+++ /dev/null
@@ -1,35 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.stream.flume;
-
-/**
- * Configuration strings for sending data to cache. Used in Flume agent
configuration file.
- */
-public class IgniteSinkConstants {
- /** Ignite configuration file path. */
- public static final String CFG_PATH = "igniteCfg";
-
- /** Cache name. */
- public static final String CFG_CACHE_NAME = "cacheName";
-
- /** Event transformer implementation. */
- public static final String CFG_EVENT_TRANSFORMER = "eventTransformer";
-
- /** Batch size. */
- public static final String CFG_BATCH_SIZE = "batchSize";
-}
diff --git
a/modules/flume/src/main/java/org/apache/ignite/stream/flume/package-info.java
b/modules/flume/src/main/java/org/apache/ignite/stream/flume/package-info.java
deleted file mode 100644
index 86833b2..0000000
---
a/modules/flume/src/main/java/org/apache/ignite/stream/flume/package-info.java
+++ /dev/null
@@ -1,22 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-/**
- * Contains implementation of Flume Sink.
- */
-
-package org.apache.ignite.stream.flume;
diff --git
a/modules/flume/src/test/java/org/apache/ignite/stream/flume/IgniteSinkTest.java
b/modules/flume/src/test/java/org/apache/ignite/stream/flume/IgniteSinkTest.java
deleted file mode 100644
index 32185df..0000000
---
a/modules/flume/src/test/java/org/apache/ignite/stream/flume/IgniteSinkTest.java
+++ /dev/null
@@ -1,139 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.stream.flume;
-
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-import org.apache.flume.Channel;
-import org.apache.flume.Context;
-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.event.EventBuilder;
-import org.apache.ignite.Ignite;
-import org.apache.ignite.IgniteCache;
-import org.apache.ignite.cache.CachePeekMode;
-import org.apache.ignite.configuration.IgniteConfiguration;
-import org.apache.ignite.events.Event;
-import org.apache.ignite.lang.IgnitePredicate;
-import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
-import org.junit.Test;
-
-import static org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_PUT;
-
-/**
- * {@link IgniteSink} test.
- */
-public class IgniteSinkTest extends GridCommonAbstractTest {
- /** Number of events to be sent to memory channel. */
- private static final int EVENT_CNT = 10000;
-
- /** Cache name. */
- private static final String CACHE_NAME = "testCache";
-
- /**
- * @throws Exception {@link Exception}.
- */
- @Test
- public void testSink() throws Exception {
- IgniteConfiguration cfg =
loadConfiguration("modules/flume/src/test/resources/example-ignite.xml");
-
- cfg.setClientMode(false);
-
- final Ignite grid = startGrid("igniteServerNode", cfg);
-
- Context channelContext = new Context();
-
- channelContext.put("capacity", String.valueOf(EVENT_CNT));
- channelContext.put("transactionCapacity", String.valueOf(EVENT_CNT));
-
- Channel memoryChannel = new MemoryChannel();
-
- Configurables.configure(memoryChannel, channelContext);
-
- final CountDownLatch latch = new CountDownLatch(EVENT_CNT);
-
- final IgnitePredicate<Event> putLsnr = new IgnitePredicate<Event>() {
- @Override public boolean apply(Event evt) {
- assert evt != null;
-
- latch.countDown();
-
- return true;
- }
- };
-
- IgniteSink sink = new IgniteSink() {
- // Setting the listener on cache before sink processing starts.
- @Override public synchronized void start() {
- super.start();
-
-
grid.events(grid.cluster().forCacheNodes(CACHE_NAME)).localListen(putLsnr,
EVT_CACHE_OBJECT_PUT);
- }
- };
-
- sink.setName("IgniteSink");
- sink.setChannel(memoryChannel);
-
- Context ctx = new Context();
-
- ctx.put(IgniteSinkConstants.CFG_CACHE_NAME, CACHE_NAME);
- ctx.put(IgniteSinkConstants.CFG_PATH, "example-ignite.xml");
- ctx.put(IgniteSinkConstants.CFG_EVENT_TRANSFORMER,
"org.apache.ignite.stream.flume.TestEventTransformer");
-
- Configurables.configure(sink, ctx);
-
- sink.start();
-
- try {
- Transaction tx = memoryChannel.getTransaction();
-
- tx.begin();
-
- for (int i = 0; i < EVENT_CNT; i++)
- memoryChannel.put(EventBuilder.withBody((String.valueOf(i) +
": " + i).getBytes()));
-
- tx.commit();
- tx.close();
-
- Sink.Status status = Sink.Status.READY;
-
- while (status != Sink.Status.BACKOFF) {
- status = sink.process();
- }
- }
- finally {
- sink.stop();
- }
-
- // Checks that 10000 events successfully processed in 10 seconds.
- assertTrue(latch.await(10, TimeUnit.SECONDS));
-
-
grid.events(grid.cluster().forCacheNodes(CACHE_NAME)).stopLocalListen(putLsnr);
-
- IgniteCache<String, Integer> cache = grid.cache(CACHE_NAME);
-
- // Checks that each event was processed properly.
- for (int i = 0; i < EVENT_CNT; i++) {
- assertEquals(i, (int)cache.get(String.valueOf(i)));
- }
-
- assertEquals(EVENT_CNT, cache.size(CachePeekMode.PRIMARY));
- }
-}
diff --git
a/modules/flume/src/test/java/org/apache/ignite/stream/flume/IgniteSinkTestSuite.java
b/modules/flume/src/test/java/org/apache/ignite/stream/flume/IgniteSinkTestSuite.java
deleted file mode 100644
index 300777d..0000000
---
a/modules/flume/src/test/java/org/apache/ignite/stream/flume/IgniteSinkTestSuite.java
+++ /dev/null
@@ -1,29 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.stream.flume;
-
-import org.junit.runner.RunWith;
-import org.junit.runners.Suite;
-
-/**
- * Tests for a Flume sink for Ignite.
- */
-@RunWith(Suite.class)
[email protected]({IgniteSinkTest.class})
-public class IgniteSinkTestSuite {
-}
diff --git
a/modules/flume/src/test/java/org/apache/ignite/stream/flume/TestEventTransformer.java
b/modules/flume/src/test/java/org/apache/ignite/stream/flume/TestEventTransformer.java
deleted file mode 100644
index c15efbf..0000000
---
a/modules/flume/src/test/java/org/apache/ignite/stream/flume/TestEventTransformer.java
+++ /dev/null
@@ -1,66 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.stream.flume;
-
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import org.apache.flume.Event;
-import org.jetbrains.annotations.Nullable;
-
-/**
- * A test transformer to convert {@link org.apache.flume.Event} to cacheable
entries.
- */
-public class TestEventTransformer implements EventTransformer<Event, String,
Integer> {
-
- /**
- * Transforms a Flume event to cacheable entries.
- *
- * @param event Flume event to transform.
- * @return Map of cacheable entries.
- */
- private Map<String, Integer> transform(Event event) {
- final Map<String, Integer> map = new HashMap<>();
-
- String eventStr = new String(event.getBody());
-
- if (!eventStr.isEmpty()) {
- String[] tokens = eventStr.split(":"); // Expects column-delimited
one line.
-
- map.put(tokens[0].trim(), Integer.valueOf(tokens[1].trim()));
- }
-
- return map;
- }
-
- /**
- * Transforms a list of Flume event to cacheable entries.
- *
- * @param events Flume events to transform.
- * @return Map of cacheable entries.
- */
- @Nullable @Override public Map<String, Integer> transform(List<Event>
events) {
- final Map<String, Integer> map = new HashMap<>();
-
- for (Event event : events) {
- map.putAll(transform(event));
- }
-
- return map;
- }
-}
diff --git
a/modules/flume/src/test/java/org/apache/ignite/stream/flume/package-info.java
b/modules/flume/src/test/java/org/apache/ignite/stream/flume/package-info.java
deleted file mode 100644
index 86833b2..0000000
---
a/modules/flume/src/test/java/org/apache/ignite/stream/flume/package-info.java
+++ /dev/null
@@ -1,22 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-/**
- * Contains implementation of Flume Sink.
- */
-
-package org.apache.ignite.stream.flume;
diff --git a/modules/flume/src/test/resources/example-ignite.xml
b/modules/flume/src/test/resources/example-ignite.xml
deleted file mode 100644
index fbb05d3..0000000
--- a/modules/flume/src/test/resources/example-ignite.xml
+++ /dev/null
@@ -1,71 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-
-<!--
- Licensed to the Apache Software Foundation (ASF) under one or more
- contributor license agreements. See the NOTICE file distributed with
- this work for additional information regarding copyright ownership.
- The ASF licenses this file to You under the Apache License, Version 2.0
- (the "License"); you may not use this file except in compliance with
- the License. You may obtain a copy of the License at
-
- http://www.apache.org/licenses/LICENSE-2.0
-
- Unless required by applicable law or agreed to in writing, software
- distributed under the License is distributed on an "AS IS" BASIS,
- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- See the License for the specific language governing permissions and
- limitations under the License.
--->
-
-<!--
- Ignite configuration with all defaults and enabled p2p deployment and
enabled events.
- Used for testing IgniteSink running Ignite in a client mode.
--->
-<beans xmlns="http://www.springframework.org/schema/beans";
- xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
- xmlns:util="http://www.springframework.org/schema/util";
- xsi:schemaLocation="
- http://www.springframework.org/schema/beans
- http://www.springframework.org/schema/beans/spring-beans.xsd
- http://www.springframework.org/schema/util
- http://www.springframework.org/schema/util/spring-util.xsd";>
- <bean id="ignite.cfg"
class="org.apache.ignite.configuration.IgniteConfiguration">
- <!-- Enable client mode. -->
- <property name="clientMode" value="true"/>
-
- <!-- Cache accessed from IgniteSink. -->
- <property name="cacheConfiguration">
- <list>
- <!-- Partitioned cache example configuration with
configurations adjusted to server nodes'. -->
- <bean
class="org.apache.ignite.configuration.CacheConfiguration">
- <property name="atomicityMode" value="ATOMIC"/>
-
- <property name="name" value="testCache"/>
- </bean>
- </list>
- </property>
-
- <!-- Enable cache events. -->
- <property name="includeEventTypes">
- <list>
- <!-- Cache events (only EVT_CACHE_OBJECT_PUT for tests). -->
- <util:constant
static-field="org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_PUT"/>
- </list>
- </property>
-
- <!-- Explicitly configure TCP discovery SPI to provide list of initial
nodes. -->
- <property name="discoverySpi">
- <bean class="org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi">
- <property name="ipFinder">
- <bean
class="org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder">
- <property name="addresses">
- <list>
- <value>127.0.0.1:47500</value>
- </list>
- </property>
- </bean>
- </property>
- </bean>
- </property>
- </bean>
-</beans>
diff --git a/parent/pom.xml b/parent/pom.xml
index c0ec1c3..423c28d 100644
--- a/parent/pom.xml
+++ b/parent/pom.xml
@@ -67,7 +67,6 @@
<easymock.version>3.4</easymock.version>
<ezmorph.bundle.version>1.0.6_1</ezmorph.bundle.version>
<ezmorph.version>1.0.6</ezmorph.version>
- <flume.ng.version>1.7.0</flume.ng.version>
<guava.retrying.version>2.0.0</guava.retrying.version>
<guava.version>25.1-jre</guava.version>
<guava14.version>14.0.1</guava14.version>
diff --git a/pom.xml b/pom.xml
index de64bce..7ff8fe4 100644
--- a/pom.xml
+++ b/pom.xml
@@ -74,7 +74,6 @@
<module>modules/cloud</module>
<module>modules/mesos</module>
<module>modules/kafka</module>
- <module>modules/flume</module>
<module>modules/yarn</module>
<module>modules/jms11</module>
<module>modules/twitter</module>
