You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flume.apache.org by rg...@apache.org on 2022/03/27 16:55:29 UTC
[flume] branch trunk updated (b155e20 -> d357e38)
This is an automated email from the ASF dual-hosted git repository.
rgoers pushed a change to branch trunk
in repository https://gitbox.apache.org/repos/asf/flume.git.
from b155e20 Do not perform RAT checks on yaml or json files
add 7392c29 FLUME-3409: httpclient security issue
add 3a7bc9c Update pom.xml
new abba25a FLUME-3413 3412 - Add LoadBalancingChannelSelector and Initializable
new c5c3c81 FLUME-3410 Upgrade HBase client libraries
new 2accecd NOOP rerun github actions
new a056876 FLUME-3151 upgrade Hadoop to 2.10.1
new f74a5ce NOOP rerun github actions
new c405e76 NOOP rerun github actions
new 4fe8ea2 Create codeql-analysis.yml
new b7975d3 Do not perform RAT checks on yaml or json files
new d357e38 Merge branch 'pjfanning-patch-2' into trunk
The 9 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
pom.xml | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
[flume] 02/09: FLUME-3410 Upgrade HBase client libraries
Posted by rg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rgoers pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/flume.git
commit c5c3c8159bf63538710227ef91385b6f604c19f2
Author: Sean Busbey <sb...@apple.com>
AuthorDate: Mon Mar 21 00:32:09 2022 -0500
FLUME-3410 Upgrade HBase client libraries
- upgrade AsyncHBase to 1.8.2 (minor version)
- upgrade HBase 1 to 1.7.0 (minor version)
- upgrade HBase 2 to 2.4.11 (patch version)
---
flume-ng-sinks/flume-ng-hbase2-sink/pom.xml | 21 ---------------------
pom.xml | 7 +++----
2 files changed, 3 insertions(+), 25 deletions(-)
diff --git a/flume-ng-sinks/flume-ng-hbase2-sink/pom.xml b/flume-ng-sinks/flume-ng-hbase2-sink/pom.xml
index 0a445a6..a6520ad 100644
--- a/flume-ng-sinks/flume-ng-hbase2-sink/pom.xml
+++ b/flume-ng-sinks/flume-ng-hbase2-sink/pom.xml
@@ -93,27 +93,6 @@
<version>${hbase2.version}</version>
</dependency>
- <!-- Jetty version needed by HBase2 tests -->
- <dependency>
- <groupId>org.eclipse.jetty</groupId>
- <artifactId>jetty-servlet</artifactId>
- <version>${hbase2.jetty.version}</version>
- <scope>test</scope>
- </dependency>
-
- <dependency>
- <groupId>org.eclipse.jetty</groupId>
- <artifactId>jetty-util</artifactId>
- <version>${hbase2.jetty.version}</version>
- <scope>test</scope>
- </dependency>
-
- <dependency>
- <groupId>org.eclipse.jetty</groupId>
- <artifactId>jetty-server</artifactId>
- <version>${hbase2.jetty.version}</version>
- <scope>test</scope>
- </dependency>
</dependencies>
</dependencyManagement>
<dependencies>
diff --git a/pom.xml b/pom.xml
index 5171bc4..dac428e 100644
--- a/pom.xml
+++ b/pom.xml
@@ -47,7 +47,7 @@ limitations under the License.
<stagingDirectory>${project.basedir}/target/docs</stagingDirectory>
<activemq.version>5.7.0</activemq.version>
- <asynchbase.version>1.7.0</asynchbase.version>
+ <asynchbase.version>1.8.2</asynchbase.version>
<avro.version>1.11.0</avro.version>
<bundle-plugin.version>2.3.7</bundle-plugin.version>
<checkstyle.tool.version>8.45.1</checkstyle.tool.version>
@@ -70,9 +70,8 @@ limitations under the License.
<guava.version>18.0</guava.version>
<guava-old.version>11.0.2</guava-old.version>
<hadoop.version>2.9.0</hadoop.version>
- <hbase.version>1.6.0</hbase.version>
- <hbase2.jetty.version>9.3.29.v20201019</hbase2.jetty.version>
- <hbase2.version>2.4.1</hbase2.version>
+ <hbase.version>1.7.1</hbase.version>
+ <hbase2.version>2.4.11</hbase2.version>
<hive.version>1.0.0</hive.version>
<httpcore.version>4.4.15</httpcore.version>
<httpclient.version>4.5.13</httpclient.version>
[flume] 04/09: FLUME-3151 upgrade Hadoop to 2.10.1
Posted by rg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rgoers pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/flume.git
commit a05687685f4b638474dc44798a4e90159b65b300
Author: Sean Busbey <sb...@apple.com>
AuthorDate: Sun Mar 20 23:34:01 2022 -0500
FLUME-3151 upgrade Hadoop to 2.10.1
---
pom.xml | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/pom.xml b/pom.xml
index dac428e..1697e82 100644
--- a/pom.xml
+++ b/pom.xml
@@ -69,7 +69,7 @@ limitations under the License.
<gson.version>2.2.2</gson.version>
<guava.version>18.0</guava.version>
<guava-old.version>11.0.2</guava-old.version>
- <hadoop.version>2.9.0</hadoop.version>
+ <hadoop.version>2.10.1</hadoop.version>
<hbase.version>1.7.1</hbase.version>
<hbase2.version>2.4.11</hbase2.version>
<hive.version>1.0.0</hive.version>
[flume] 09/09: Merge branch 'pjfanning-patch-2' into trunk
Posted by rg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rgoers pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/flume.git
commit d357e3886dad8bed6bfca553a62c51697417f0e6
Merge: b155e20 b7975d3
Author: Ralph Goers <rg...@apache.org>
AuthorDate: Sun Mar 27 09:55:08 2022 -0700
Merge branch 'pjfanning-patch-2' into trunk
pom.xml | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
[flume] 07/09: Create codeql-analysis.yml
Posted by rg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rgoers pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/flume.git
commit 4fe8ea24cc0403bd5301cef1054da31e78c18d6f
Author: Ralph Goers <rg...@apache.org>
AuthorDate: Sat Mar 26 23:18:12 2022 -0700
Create codeql-analysis.yml
---
.github/workflows/codeql-analysis.yml | 70 +++++++++++++++++++++++++++++++++++
1 file changed, 70 insertions(+)
diff --git a/.github/workflows/codeql-analysis.yml b/.github/workflows/codeql-analysis.yml
new file mode 100644
index 0000000..20d8be9
--- /dev/null
+++ b/.github/workflows/codeql-analysis.yml
@@ -0,0 +1,70 @@
+# For most projects, this workflow file will not need changing; you simply need
+# to commit it to your repository.
+#
+# You may wish to alter this file to override the set of languages analyzed,
+# or to provide custom queries or build logic.
+#
+# ******** NOTE ********
+# We have attempted to detect the languages in your repository. Please check
+# the `language` matrix defined below to confirm you have the correct set of
+# supported CodeQL languages.
+#
+name: "CodeQL"
+
+on:
+ push:
+ branches: [ trunk ]
+ pull_request:
+ # The branches below must be a subset of the branches above
+ branches: [ trunk ]
+ schedule:
+ - cron: '15 0 * * 5'
+
+jobs:
+ analyze:
+ name: Analyze
+ runs-on: ubuntu-latest
+ permissions:
+ actions: read
+ contents: read
+ security-events: write
+
+ strategy:
+ fail-fast: false
+ matrix:
+ language: [ 'java', 'python' ]
+ # CodeQL supports [ 'cpp', 'csharp', 'go', 'java', 'javascript', 'python', 'ruby' ]
+ # Learn more about CodeQL language support at https://git.io/codeql-language-support
+
+ steps:
+ - name: Checkout repository
+ uses: actions/checkout@v2
+
+ # Initializes the CodeQL tools for scanning.
+ - name: Initialize CodeQL
+ uses: github/codeql-action/init@v1
+ with:
+ languages: ${{ matrix.language }}
+ # If you wish to specify custom queries, you can do so here or in a config file.
+ # By default, queries listed here will override any specified in a config file.
+ # Prefix the list here with "+" to use these queries and those in the config file.
+ # queries: ./path/to/local/query, your-org/your-repo/queries@main
+
+ # Autobuild attempts to build any compiled languages (C/C++, C#, or Java).
+ # If this step fails, then you should remove it and run the build manually (see below)
+ - name: Autobuild
+ uses: github/codeql-action/autobuild@v1
+
+ # âšī¸ Command-line programs to run using the OS shell.
+ # đ https://git.io/JvXDl
+
+ # âī¸ If the Autobuild fails above, remove it and uncomment the following three lines
+ # and modify them (or add more) to build your code if your project
+ # uses a compiled language
+
+ #- run: |
+ # make bootstrap
+ # make release
+
+ - name: Perform CodeQL Analysis
+ uses: github/codeql-action/analyze@v1
[flume] 06/09: NOOP rerun github actions
Posted by rg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rgoers pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/flume.git
commit c405e76e74fa89fc807c66ba5e3a1dcaabd44076
Author: Sean Busbey <sb...@apple.com>
AuthorDate: Tue Mar 22 08:51:28 2022 -0500
NOOP rerun github actions
[flume] 08/09: Do not perform RAT checks on yaml or json files
Posted by rg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rgoers pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/flume.git
commit b7975d3b9ea33064e8a5da5127d0a22bb6d3b797
Author: Ralph Goers <rg...@apache.org>
AuthorDate: Sat Mar 26 23:27:31 2022 -0700
Do not perform RAT checks on yaml or json files
---
pom.xml | 3 +++
1 file changed, 3 insertions(+)
diff --git a/pom.xml b/pom.xml
index 1697e82..9c47c25 100644
--- a/pom.xml
+++ b/pom.xml
@@ -531,6 +531,9 @@ limitations under the License.
<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>
[flume] 05/09: NOOP rerun github actions
Posted by rg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rgoers pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/flume.git
commit f74a5cef1b2abb581363791e68be2f177df0ffb7
Author: Sean Busbey <sb...@apple.com>
AuthorDate: Tue Mar 22 08:52:37 2022 -0500
NOOP rerun github actions
[flume] 03/09: NOOP rerun github actions
Posted by rg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rgoers pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/flume.git
commit 2accecd66656e3e2232d81a809e7cdd54813d9b5
Author: Sean Busbey <sb...@apple.com>
AuthorDate: Tue Mar 22 08:51:28 2022 -0500
NOOP rerun github actions
[flume] 01/09: FLUME-3413 3412 - Add LoadBalancingChannelSelector and Initializable
Posted by rg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rgoers pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/flume.git
commit abba25ac7f1a0be5669a780dadc0078903ad8793
Author: Ralph Goers <rg...@apache.org>
AuthorDate: Sat Mar 26 12:48:21 2022 -0700
FLUME-3413 3412 - Add LoadBalancingChannelSelector and Initializable
---
.mvn/local-settings.xml | 7 -
.../flume/conf/channel/ChannelSelectorType.java | 5 +
.../channel/LoadBalancingChannelSelector.java | 152 +++++++++++++++++++++
.../flume/sink/AbstractSingleSinkProcessor.java | 65 +++++++++
.../apache/flume/sink/AbstractSinkProcessor.java | 2 +-
.../apache/flume/sink/DefaultSinkProcessor.java | 39 +-----
.../main/java/org/apache/flume/sink/NullSink.java | 8 ++
.../channel/TestLoadBalancingChannelSelector.java | 72 ++++++++++
flume-ng-doc/sphinx/FlumeDeveloperGuide.rst | 90 ++++++++++++
flume-ng-doc/sphinx/FlumeUserGuide.rst | 23 ++++
flume-ng-node/pom.xml | 12 ++
.../java/org/apache/flume/node/Application.java | 36 +++++
.../java/org/apache/flume/node/Initializable.java | 30 ++++
.../org/apache/flume/node/TestApplication.java | 67 +++++++++
.../java/org/apache/flume/sink/NullInitSink.java | 98 ++++++-------
.../org/apache/flume/source/EventProcessor.java | 33 +++++
.../java/org/apache/flume/source/LocalSource.java | 89 ++++++++++++
.../src/test/resources/flume-conf-init.properties | 51 +++++++
flume-ng-node/src/test/resources/log4j2.xml | 4 +-
19 files changed, 778 insertions(+), 105 deletions(-)
diff --git a/.mvn/local-settings.xml b/.mvn/local-settings.xml
index 6bf134e..55d11d3 100644
--- a/.mvn/local-settings.xml
+++ b/.mvn/local-settings.xml
@@ -7,49 +7,42 @@
<mirrorOf>wso2</mirrorOf>
<name>wso2-unblocked</name>
<url>http://dist.wso2.org/maven2/</url>
- <blocked>false</blocked>
</mirror>
<mirror>
<id>maven-twttr-unblocked</id>
<mirrorOf>maven-twttr</mirrorOf>
<name>maven-twttr-unblocked</name>
<url>http://maven.twttr.com</url>
- <blocked>false</blocked>
</mirror>
<mirror>
<id>maven-restlet-unblocked</id>
<mirrorOf>maven-restlet</mirrorOf>
<name>maven-restlet-unblocked</name>
<url>http://maven.restlet.org</url>
- <blocked>false</blocked>
</mirror>
<mirror>
<id>datanucleus-unblocked</id>
<mirrorOf>datanucleus</mirrorOf>
<name>datanucleus-unblocked</name>
<url>http://www.datanucleus.org/downloads/maven2</url>
- <blocked>false</blocked>
</mirror>
<mirror>
<id>glassfish-repository-unblocked</id>
<mirrorOf>glassfish-repository</mirrorOf>
<name>glassfish-repository-unblocked</name>
<url>http://maven.glassfish.org/content/groups/glassfish</url>
- <blocked>false</blocked>
</mirror>
<mirror>
<id>glassfish-repo-archive-unblocked</id>
<mirrorOf>glassfish-repo-archive</mirrorOf>
<name>glassfish-repo-archive-unblocked</name>
<url>http://maven.glassfish.org/content/groups/glassfish</url>
- <blocked>false</blocked>
</mirror>
<mirror>
<id>conjars-unblocked</id>
<mirrorOf>conjars</mirrorOf>
<name>conjars-unblocked</name>
<url>http://conjars.org/repo</url>
- <blocked>false</blocked>
</mirror>
</mirrors>
</settings>
\ No newline at end of file
diff --git a/flume-ng-configuration/src/main/java/org/apache/flume/conf/channel/ChannelSelectorType.java b/flume-ng-configuration/src/main/java/org/apache/flume/conf/channel/ChannelSelectorType.java
index d1a0dc2..e329ec3 100644
--- a/flume-ng-configuration/src/main/java/org/apache/flume/conf/channel/ChannelSelectorType.java
+++ b/flume-ng-configuration/src/main/java/org/apache/flume/conf/channel/ChannelSelectorType.java
@@ -34,6 +34,11 @@ public enum ChannelSelectorType implements ComponentWithClassName {
REPLICATING("org.apache.flume.channel.ReplicatingChannelSelector"),
/**
+ * Load balancing channel selector.
+ */
+ LOAD_BALANCING("org.apache.flume.channel.LoadBalancingChannelSelector"),
+
+ /**
* Multiplexing channel selector.
*/
MULTIPLEXING("org.apache.flume.channel.MultiplexingChannelSelector");
diff --git a/flume-ng-core/src/main/java/org/apache/flume/channel/LoadBalancingChannelSelector.java b/flume-ng-core/src/main/java/org/apache/flume/channel/LoadBalancingChannelSelector.java
new file mode 100644
index 0000000..0b080b2
--- /dev/null
+++ b/flume-ng-core/src/main/java/org/apache/flume/channel/LoadBalancingChannelSelector.java
@@ -0,0 +1,152 @@
+/*
+ * 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.channel;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Random;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.flume.Channel;
+import org.apache.flume.Context;
+import org.apache.flume.Event;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+
+/**
+ * Load balancing channel selector. This selector allows for load balancing
+ * between channels based on various policy configuration options. This serves a similar purpose
+ * to the LoadBalancingSinkProcessor except it allows the sinks to run in their own threads instead
+ * of in just one.
+ *
+ * <p>The <tt>LoadBalancingChannelSelector</tt> maintains an indexed list of
+ * active channels on which the load must be distributed. This implementation
+ * supports distributing load using either via <tt>ROUND_ROBIN</tt> or via
+ * <tt>RANDOM</tt> selection mechanism. The choice of selection mechanism
+ * defaults to <tt>ROUND_ROBIN</tt> type, but can be overridden via
+ * configuration.</p>
+ */
+public class LoadBalancingChannelSelector extends AbstractChannelSelector {
+ private final List<Channel> emptyList = Collections.emptyList();
+ private ChannelPicker picker;
+
+ @Override
+ public List<Channel> getRequiredChannels(Event event) {
+ Channel ch = picker.getChannel();
+ Preconditions.checkNotNull(ch, "Channel picker returned null");
+ return Lists.newArrayList(ch);
+ }
+
+ @Override
+ public List<Channel> getOptionalChannels(Event event) {
+ return emptyList;
+ }
+
+ @Override
+ public void configure(Context context) {
+ List<Channel> channels = getAllChannels();
+ String strPolicy = context.getString("policy", Policy.ROUND_ROBIN.toString());
+ Policy policy;
+ // instantiate policy
+ try {
+ policy = Policy.valueOf(strPolicy.toUpperCase());
+ } catch (IllegalArgumentException ex) {
+ throw new IllegalArgumentException("Invalid policy: " + strPolicy, ex);
+ }
+ // instantiate picker
+ try {
+ picker = policy.getPolicyClass().newInstance();
+ picker.setChannels(channels);
+ } catch (InstantiationException | IllegalAccessException ex) {
+ throw new IllegalArgumentException("Cannot instantiate policy class from policy enum "
+ + policy, ex);
+ }
+ }
+
+ /**
+ * Definitions for the various policy types
+ */
+ private enum Policy {
+ ROUND_ROBIN(RoundRobinPolicy.class),
+ RANDOM(RandomPolicy.class);
+
+ private final Class<? extends ChannelPicker> clazz;
+
+ Policy(Class<? extends ChannelPicker> clazz) {
+ this.clazz = clazz;
+ }
+
+ public Class<? extends ChannelPicker> getPolicyClass() {
+ return clazz;
+ }
+ }
+
+ private interface ChannelPicker {
+ Channel getChannel();
+
+ void setChannels(List<Channel> channels);
+ }
+
+ /**
+ * Selects channels in a round-robin fashion
+ */
+ private static class RoundRobinPolicy implements ChannelPicker {
+
+ private final AtomicInteger next = new AtomicInteger(0);
+ private List<Channel> channels;
+
+ public RoundRobinPolicy() {
+ }
+
+ @Override
+ public void setChannels(List<Channel> channels) {
+ this.channels = channels;
+ }
+
+ @Override
+ public Channel getChannel() {
+ return channels.get(next.getAndAccumulate(channels.size(), (x, y) -> ++x < y ? x : 0));
+ }
+ }
+
+ /**
+ * Selects a channel at random
+ */
+ private static class RandomPolicy implements ChannelPicker {
+ private List<Channel> channels;
+ private final Random random = new Random(System.currentTimeMillis());
+
+ public RandomPolicy() {
+ }
+
+ @Override
+ public void setChannels(List<Channel> channels) {
+ this.channels = channels;
+ }
+
+ @Override
+ public Channel getChannel() {
+ int size = channels.size();
+ int pick = random.nextInt(size);
+ return channels.get(pick);
+ }
+
+ }
+}
diff --git a/flume-ng-core/src/main/java/org/apache/flume/sink/AbstractSingleSinkProcessor.java b/flume-ng-core/src/main/java/org/apache/flume/sink/AbstractSingleSinkProcessor.java
new file mode 100644
index 0000000..6b1019b
--- /dev/null
+++ b/flume-ng-core/src/main/java/org/apache/flume/sink/AbstractSingleSinkProcessor.java
@@ -0,0 +1,65 @@
+/*
+ * 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;
+
+import java.util.List;
+
+import org.apache.flume.Sink;
+import org.apache.flume.SinkProcessor;
+import org.apache.flume.lifecycle.LifecycleState;
+
+import com.google.common.base.Preconditions;
+
+/**
+ * A Sink Processor that only accesses a single Sink.
+ */
+public abstract class AbstractSingleSinkProcessor implements SinkProcessor {
+ protected Sink sink;
+ private LifecycleState lifecycleState;
+
+ @Override
+ public void start() {
+ Preconditions.checkNotNull(sink, "DefaultSinkProcessor sink not set");
+ sink.start();
+ lifecycleState = LifecycleState.START;
+ }
+
+ @Override
+ public void stop() {
+ Preconditions.checkNotNull(sink, "DefaultSinkProcessor sink not set");
+ sink.stop();
+ lifecycleState = LifecycleState.STOP;
+ }
+
+ @Override
+ public LifecycleState getLifecycleState() {
+ return lifecycleState;
+ }
+
+ @Override
+ public void setSinks(List<Sink> sinks) {
+ Preconditions.checkNotNull(sinks);
+ Preconditions.checkArgument(sinks.size() == 1, "DefaultSinkPolicy can "
+ + "only handle one sink, "
+ + "try using a policy that supports multiple sinks");
+ sink = sinks.get(0);
+ }
+
+ public Sink getSink() {
+ return sink;
+ }
+}
diff --git a/flume-ng-core/src/main/java/org/apache/flume/sink/AbstractSinkProcessor.java b/flume-ng-core/src/main/java/org/apache/flume/sink/AbstractSinkProcessor.java
index 3de653a..a8398fe 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/sink/AbstractSinkProcessor.java
+++ b/flume-ng-core/src/main/java/org/apache/flume/sink/AbstractSinkProcessor.java
@@ -64,7 +64,7 @@ public abstract class AbstractSinkProcessor implements SinkProcessor {
sinkList = Collections.unmodifiableList(list);
}
- protected List<Sink> getSinks() {
+ public List<Sink> getSinks() {
return sinkList;
}
}
diff --git a/flume-ng-core/src/main/java/org/apache/flume/sink/DefaultSinkProcessor.java b/flume-ng-core/src/main/java/org/apache/flume/sink/DefaultSinkProcessor.java
index 2da9264..b11bab7 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/sink/DefaultSinkProcessor.java
+++ b/flume-ng-core/src/main/java/org/apache/flume/sink/DefaultSinkProcessor.java
@@ -17,46 +17,18 @@
*/
package org.apache.flume.sink;
-import java.util.List;
-
import org.apache.flume.Context;
import org.apache.flume.EventDeliveryException;
-import org.apache.flume.Sink;
import org.apache.flume.Sink.Status;
-import org.apache.flume.SinkProcessor;
import org.apache.flume.conf.ComponentConfiguration;
import org.apache.flume.conf.ConfigurableComponent;
-import org.apache.flume.lifecycle.LifecycleState;
-
-import com.google.common.base.Preconditions;
/**
* Default sink processor that only accepts a single sink, passing on process
* results without any additional handling. Suitable for all sinks that aren't
* assigned to a group.
*/
-public class DefaultSinkProcessor implements SinkProcessor, ConfigurableComponent {
- private Sink sink;
- private LifecycleState lifecycleState;
-
- @Override
- public void start() {
- Preconditions.checkNotNull(sink, "DefaultSinkProcessor sink not set");
- sink.start();
- lifecycleState = LifecycleState.START;
- }
-
- @Override
- public void stop() {
- Preconditions.checkNotNull(sink, "DefaultSinkProcessor sink not set");
- sink.stop();
- lifecycleState = LifecycleState.STOP;
- }
-
- @Override
- public LifecycleState getLifecycleState() {
- return lifecycleState;
- }
+public class DefaultSinkProcessor extends AbstractSingleSinkProcessor implements ConfigurableComponent {
@Override
public void configure(Context context) {
@@ -68,15 +40,6 @@ public class DefaultSinkProcessor implements SinkProcessor, ConfigurableComponen
}
@Override
- public void setSinks(List<Sink> sinks) {
- Preconditions.checkNotNull(sinks);
- Preconditions.checkArgument(sinks.size() == 1, "DefaultSinkPolicy can "
- + "only handle one sink, "
- + "try using a policy that supports multiple sinks");
- sink = sinks.get(0);
- }
-
- @Override
public void configure(ComponentConfiguration conf) {
}
diff --git a/flume-ng-core/src/main/java/org/apache/flume/sink/NullSink.java b/flume-ng-core/src/main/java/org/apache/flume/sink/NullSink.java
index 9347f8e..9ca2b8f 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/sink/NullSink.java
+++ b/flume-ng-core/src/main/java/org/apache/flume/sink/NullSink.java
@@ -138,6 +138,14 @@ public class NullSink extends AbstractSink implements Configurable, BatchSizeSup
return "NullSink " + getName() + " { batchSize: " + batchSize + " }";
}
+ public CounterGroup getCounterGroup() {
+ return counterGroup;
+ }
+
+ public int getLogEveryNEvents() {
+ return logEveryNEvents;
+ }
+
@Override
public long getBatchSize() {
return batchSize;
diff --git a/flume-ng-core/src/test/java/org/apache/flume/channel/TestLoadBalancingChannelSelector.java b/flume-ng-core/src/test/java/org/apache/flume/channel/TestLoadBalancingChannelSelector.java
new file mode 100644
index 0000000..e0e9e79
--- /dev/null
+++ b/flume-ng-core/src/test/java/org/apache/flume/channel/TestLoadBalancingChannelSelector.java
@@ -0,0 +1,72 @@
+/*
+ * 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.channel;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.flume.Channel;
+import org.apache.flume.ChannelSelector;
+import org.apache.flume.Context;
+import org.apache.flume.conf.BasicConfigurationConstants;
+import org.apache.flume.conf.channel.ChannelSelectorType;
+import org.junit.Before;
+import org.junit.Test;
+
+import junit.framework.Assert;
+
+public class TestLoadBalancingChannelSelector {
+
+ private List<Channel> channels = new ArrayList<Channel>();
+
+ private ChannelSelector selector;
+
+ @Before
+ public void setUp() throws Exception {
+ channels.clear();
+ channels.add(MockChannel.createMockChannel("ch1"));
+ channels.add(MockChannel.createMockChannel("ch2"));
+ channels.add(MockChannel.createMockChannel("ch3"));
+ channels.add(MockChannel.createMockChannel("ch4"));
+ Map<String, String> config = new HashMap<>();
+ config.put(BasicConfigurationConstants.CONFIG_TYPE, ChannelSelectorType.LOAD_BALANCING.name());
+ selector = ChannelSelectorFactory.create(channels, config);
+ }
+
+ @Test
+ public void testLoadBalancingSelector() throws Exception {
+ selector.configure(new Context());
+ validateChannel(selector, "ch1");
+ validateChannel(selector, "ch2");
+ validateChannel(selector, "ch3");
+ validateChannel(selector, "ch4");
+
+ List<Channel> optCh = selector.getOptionalChannels(new MockEvent());
+ Assert.assertEquals(0, optCh.size());
+ }
+
+ private void validateChannel(ChannelSelector selector, String channelName) {
+ List<Channel> channels = selector.getRequiredChannels(new MockEvent());
+ Assert.assertNotNull(channels);
+ Assert.assertEquals(1, channels.size());
+ Assert.assertEquals(channelName, channels.get(0).getName());
+ }
+}
diff --git a/flume-ng-doc/sphinx/FlumeDeveloperGuide.rst b/flume-ng-doc/sphinx/FlumeDeveloperGuide.rst
index de1e92a..f58da81 100644
--- a/flume-ng-doc/sphinx/FlumeDeveloperGuide.rst
+++ b/flume-ng-doc/sphinx/FlumeDeveloperGuide.rst
@@ -866,3 +866,93 @@ Channel
~~~~~~~
TBD
+
+Initializable
+~~~~~~~~~~~~~
+
+As of Flume 1.10.0 Sources, Sinks, and Channels may implement the Intitializable interface. Doing so
+allows the component to have access the materialized configuration before any of the components have been
+started.
+
+This example shows a Sink being configured with the name of a Source. While initializing it will
+retrieve the Source from the configuration and save it. During event processing a new event will be
+sent to the Source, presumably after the event has be modified in some way.
+
+.. code-block:: java
+
+ public class NullInitSink extends NullSink implements Initializable {
+
+ private static final Logger logger = LoggerFactory.getLogger(NullInitSink.class);
+ private String sourceName = null;
+ private EventProcessor eventProcessor = null;
+ private long total = 0;
+
+ public NullInitSink() {
+ super();
+ }
+
+ @Override
+ public void configure(Context context) {
+ sourceName = context.getString("targetSource");
+ super.configure(context);
+
+ }
+
+ @Override
+ public void initialize(MaterializedConfiguration configuration) {
+ logger.debug("Locating source for event publishing");
+ for (Map.Entry<String, SourceRunner> entry : configuration.getSourceRunners().entrySet()) {
+ if (entry.getKey().equals(sourceName)) {
+ Source source = entry.getValue().getSource();
+ if (source instanceof EventProcessor) {
+ eventProcessor = (EventProcessor) source;
+ logger.debug("Found event processor {}", source.getName());
+ return;
+ }
+ }
+ }
+ logger.warn("No Source named {} found for republishing events.", sourceName);
+ }
+
+ @Override
+ public Status process() throws EventDeliveryException {
+ Status status = Status.READY;
+
+ Channel channel = getChannel();
+ Transaction transaction = channel.getTransaction();
+ Event event = null;
+ CounterGroup counterGroup = getCounterGroup();
+ long batchSize = getBatchSize();
+ long eventCounter = counterGroup.get("events.success");
+
+ try {
+ transaction.begin();
+ int i = 0;
+ for (i = 0; i < batchSize; i++) {
+ event = channel.take();
+ if (event != null) {
+ long id = Long.parseLong(new String(event.getBody()));
+ total += id;
+ event.getHeaders().put("Total", Long.toString(total));
+ eventProcessor.processEvent(event);
+ logger.info("Null sink {} successful processed event {}", getName(), id);
+ } else {
+ status = Status.BACKOFF;
+ break;
+ }
+ }
+ transaction.commit();
+ counterGroup.addAndGet("events.success", (long) Math.min(batchSize, i));
+ counterGroup.incrementAndGet("transaction.success");
+ } catch (Exception ex) {
+ transaction.rollback();
+ counterGroup.incrementAndGet("transaction.failed");
+ logger.error("Failed to deliver event. Exception follows.", ex);
+ throw new EventDeliveryException("Failed to deliver event: " + event, ex);
+ } finally {
+ transaction.close();
+ }
+
+ return status;
+ }
+ }
\ No newline at end of file
diff --git a/flume-ng-doc/sphinx/FlumeUserGuide.rst b/flume-ng-doc/sphinx/FlumeUserGuide.rst
index 315e5c4..eaa73b7 100644
--- a/flume-ng-doc/sphinx/FlumeUserGuide.rst
+++ b/flume-ng-doc/sphinx/FlumeUserGuide.rst
@@ -4121,6 +4121,29 @@ In the above configuration, c3 is an optional channel. Failure to write to c3 is
simply ignored. Since c1 and c2 are not marked optional, failure to write to
those channels will cause the transaction to fail.
+Load Balancing Channel Selector
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+Load balancing channel selector provides the ability to load-balance flow over multiple channels. This
+effectively allows the incoming data to be processed on multiple threads. It maintains an indexed list of active channels on which the load must be distributed. Implementation supports distributing load using either via round_robin or random selection mechanisms. The choice of selection mechanism defaults to round_robin type, but can be overridden via configuration.
+
+Required properties are in **bold**.
+
+================== ===================== =================================================
+Property Name Default Description
+================== ===================== =================================================
+selector.type replicating The component type name, needs to be ``load_balancing``
+selector.policy ``round_robin`` Selection mechanism. Must be either ``round_robin`` or ``random``.
+================== ===================== =================================================
+
+Example for agent named a1 and it's source called r1:
+
+.. code-block:: properties
+
+ a1.sources = r1
+ a1.channels = c1 c2 c3 c4
+ a1.sources.r1.channels = c1 c2 c3 c4
+ a1.sources.r1.selector.type = load_balancing
+ a1.sources.r1.selector.policy = round_robin
Multiplexing Channel Selector
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
diff --git a/flume-ng-node/pom.xml b/flume-ng-node/pom.xml
index cb6979a..17c8456 100644
--- a/flume-ng-node/pom.xml
+++ b/flume-ng-node/pom.xml
@@ -86,6 +86,18 @@
<dependency>
<groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-api</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-core</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-1.2-api</artifactId>
<scope>test</scope>
</dependency>
diff --git a/flume-ng-node/src/main/java/org/apache/flume/node/Application.java b/flume-ng-node/src/main/java/org/apache/flume/node/Application.java
index 1f4df59..e30ede8 100644
--- a/flume-ng-node/src/main/java/org/apache/flume/node/Application.java
+++ b/flume-ng-node/src/main/java/org/apache/flume/node/Application.java
@@ -43,7 +43,10 @@ import org.apache.commons.cli.ParseException;
import org.apache.commons.lang.StringUtils;
import org.apache.flume.Channel;
import org.apache.flume.Context;
+import org.apache.flume.Sink;
+import org.apache.flume.SinkProcessor;
import org.apache.flume.SinkRunner;
+import org.apache.flume.Source;
import org.apache.flume.SourceRunner;
import org.apache.flume.instrumentation.MonitorService;
import org.apache.flume.instrumentation.MonitoringType;
@@ -53,6 +56,8 @@ import org.apache.flume.lifecycle.LifecycleSupervisor;
import org.apache.flume.lifecycle.LifecycleSupervisor.SupervisorPolicy;
import org.apache.flume.node.net.AuthorizationProvider;
import org.apache.flume.node.net.BasicAuthorizationProvider;
+import org.apache.flume.sink.AbstractSingleSinkProcessor;
+import org.apache.flume.sink.AbstractSinkProcessor;
import org.apache.flume.util.SSLUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -104,6 +109,7 @@ public class Application {
try {
lifecycleLock.lockInterruptibly();
stopAllComponents();
+ initializeAllComponents(conf);
startAllComponents(conf);
} catch (InterruptedException e) {
logger.info("Interrupted while trying to handle configuration event");
@@ -167,6 +173,36 @@ public class Application {
}
}
+ private void initializeAllComponents(MaterializedConfiguration materializedConfiguration) {
+ logger.info("Initializing components");
+ for (Channel ch : materializedConfiguration.getChannels().values()) {
+ while (ch.getLifecycleState() != LifecycleState.START && ch instanceof Initializable) {
+ ((Initializable) ch).initialize(materializedConfiguration);
+ }
+ }
+ for (SinkRunner sinkRunner : materializedConfiguration.getSinkRunners().values()) {
+ SinkProcessor processor = sinkRunner.getPolicy();
+ if (processor instanceof AbstractSingleSinkProcessor) {
+ Sink sink = ((AbstractSingleSinkProcessor) processor).getSink();
+ if (sink instanceof Initializable) {
+ ((Initializable) sink).initialize(materializedConfiguration);
+ }
+ } else if (processor instanceof AbstractSinkProcessor) {
+ for (Sink sink : ((AbstractSinkProcessor) processor).getSinks()) {
+ if (sink instanceof Initializable) {
+ ((Initializable) sink).initialize(materializedConfiguration);
+ }
+ }
+ }
+ }
+ for (SourceRunner sourceRunner : materializedConfiguration.getSourceRunners().values()) {
+ Source source = sourceRunner.getSource();
+ if (source instanceof Initializable) {
+ ((Initializable) source).initialize(materializedConfiguration);
+ }
+ }
+ }
+
private void startAllComponents(MaterializedConfiguration materializedConfiguration) {
logger.info("Starting new configuration:{}", materializedConfiguration);
diff --git a/flume-ng-node/src/main/java/org/apache/flume/node/Initializable.java b/flume-ng-node/src/main/java/org/apache/flume/node/Initializable.java
new file mode 100644
index 0000000..114990e
--- /dev/null
+++ b/flume-ng-node/src/main/java/org/apache/flume/node/Initializable.java
@@ -0,0 +1,30 @@
+/*
+ * 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.node;
+
+/**
+ * An interface implmmented by components that need access after all components have been created but before
+ * any have been started.
+ */
+public interface Initializable {
+
+ /**
+ * Called to initialize the component.
+ * @param configuration the materialized configuration.
+ */
+ void initialize(MaterializedConfiguration configuration);
+}
diff --git a/flume-ng-node/src/test/java/org/apache/flume/node/TestApplication.java b/flume-ng-node/src/test/java/org/apache/flume/node/TestApplication.java
index 3853d50..00d15c5 100644
--- a/flume-ng-node/src/test/java/org/apache/flume/node/TestApplication.java
+++ b/flume-ng-node/src/test/java/org/apache/flume/node/TestApplication.java
@@ -22,17 +22,25 @@ package org.apache.flume.node;
import static org.mockito.Mockito.*;
import java.io.File;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
import java.util.List;
+import java.util.Map;
import java.util.Random;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.commons.io.FileUtils;
import org.apache.flume.Channel;
+import org.apache.flume.Event;
import org.apache.flume.SinkRunner;
import org.apache.flume.SourceRunner;
+import org.apache.flume.Transaction;
+import org.apache.flume.event.SimpleEvent;
import org.apache.flume.lifecycle.LifecycleAware;
import org.apache.flume.lifecycle.LifecycleState;
+import org.apache.flume.source.EventProcessor;
import org.junit.After;
+import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.mockito.invocation.InvocationOnMock;
@@ -195,4 +203,63 @@ public class TestApplication {
Thread.sleep(1500L);
application.stop();
}
+
+ @Test
+ public void testFlumeInit() throws Exception {
+ File configFile = new File(baseDir, "flume-conf-init.properties");
+ Files.copy(new File(getClass().getClassLoader()
+ .getResource("flume-conf-init.properties").getFile()), configFile);
+ ConfigurationSource source = new FileConfigurationSource(configFile.toURI());
+ List<ConfigurationSource> sourceList = new ArrayList<>();
+ sourceList.add(source);
+ UriConfigurationProvider configurationProvider =
+ new UriConfigurationProvider("host1", sourceList, null, null, 1);
+ List<LifecycleAware> components = Lists.newArrayList();
+ Application application = new Application(components);
+ MaterializedConfiguration configuration = configurationProvider.getConfiguration();
+ Assert.assertNotNull("Unable to create configuration", configuration);
+ application.handleConfigurationEvent(configuration);
+ application.start();
+ Map<String, Channel> channels = configuration.getChannels();
+ Channel channel = channels.get("processedChannel");
+ Assert.assertNotNull("Channel not found", channel);
+ Map<String, SourceRunner> sourceRunners = configuration.getSourceRunners();
+ Assert.assertNotNull("No source runners", sourceRunners);
+ SourceRunner runner = sourceRunners.get("source1");
+ Assert.assertNotNull("No source runner", runner);
+ EventProcessor processor = (EventProcessor) runner.getSource();
+ long[] expected = new long[]{1, 3, 6, 10, 15};
+ for (int i = 0; i < 5; ++i) {
+ Event event = new SimpleEvent();
+ event.setBody(Long.toString(i + 1).getBytes(StandardCharsets.UTF_8));
+ processor.processEvent(event);
+ }
+ Thread.sleep(500);
+ for (int i = 0; i < 5; ++i) {
+ Event event = getEvent(channel);
+ Assert.assertNotNull("No event returned on iteration " + i, event);
+ String val = event.getHeaders().get("Total");
+ Assert.assertNotNull("No Total in event " + i, val);
+ long total = Long.parseLong(val);
+ Assert.assertEquals(expected[i], total);
+ }
+ application.stop();
+ }
+
+ private Event getEvent(Channel channel) {
+ Transaction transaction = channel.getTransaction();
+ Event event = null;
+
+ try {
+ transaction.begin();
+ event = channel.take();
+ transaction.commit();
+ } catch (Exception ex) {
+ transaction.rollback();
+ Assert.fail("Failed to retrieve Flume Event: " + ex.getMessage());
+ } finally {
+ transaction.close();
+ }
+ return event;
+ }
}
diff --git a/flume-ng-core/src/main/java/org/apache/flume/sink/NullSink.java b/flume-ng-node/src/test/java/org/apache/flume/sink/NullInitSink.java
similarity index 58%
copy from flume-ng-core/src/main/java/org/apache/flume/sink/NullSink.java
copy to flume-ng-node/src/test/java/org/apache/flume/sink/NullInitSink.java
index 9347f8e..00296e6 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/sink/NullSink.java
+++ b/flume-ng-node/src/test/java/org/apache/flume/sink/NullInitSink.java
@@ -17,16 +17,20 @@
*/
package org.apache.flume.sink;
-import com.google.common.base.Preconditions;
+import java.util.Map;
+
import org.apache.flume.Channel;
import org.apache.flume.Context;
import org.apache.flume.CounterGroup;
import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.Sink;
+import org.apache.flume.Source;
+import org.apache.flume.SourceRunner;
import org.apache.flume.Transaction;
-import org.apache.flume.conf.BatchSizeSupported;
-import org.apache.flume.conf.Configurable;
+import org.apache.flume.node.Initializable;
+import org.apache.flume.node.MaterializedConfiguration;
+import org.apache.flume.source.EventProcessor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -48,32 +52,38 @@ import org.slf4j.LoggerFactory;
* TODO
* </p>
*/
-public class NullSink extends AbstractSink implements Configurable, BatchSizeSupported {
-
- private static final Logger logger = LoggerFactory.getLogger(NullSink.class);
+public class NullInitSink extends NullSink implements Initializable {
- private static final int DFLT_BATCH_SIZE = 100;
- private static final int DFLT_LOG_EVERY_N_EVENTS = 10000;
+ private static final Logger logger = LoggerFactory.getLogger(NullInitSink.class);
+ private String sourceName = null;
+ private EventProcessor eventProcessor = null;
+ private long total = 0;
- private CounterGroup counterGroup;
- private int batchSize = DFLT_BATCH_SIZE;
- private int logEveryNEvents = DFLT_LOG_EVERY_N_EVENTS;
-
- public NullSink() {
- counterGroup = new CounterGroup();
+ public NullInitSink() {
+ super();
}
@Override
public void configure(Context context) {
- batchSize = context.getInteger("batchSize", DFLT_BATCH_SIZE);
- logger.debug(this.getName() + " " +
- "batch size set to " + String.valueOf(batchSize));
- Preconditions.checkArgument(batchSize > 0, "Batch size must be > 0");
+ sourceName = context.getString("targetSource");
+ super.configure(context);
- logEveryNEvents = context.getInteger("logEveryNEvents", DFLT_LOG_EVERY_N_EVENTS);
- logger.debug(this.getName() + " " +
- "log event N events set to " + logEveryNEvents);
- Preconditions.checkArgument(logEveryNEvents > 0, "logEveryNEvents must be > 0");
+ }
+
+ @Override
+ public void initialize(MaterializedConfiguration configuration) {
+ logger.debug("Locating source for event publishing");
+ for (Map.Entry<String, SourceRunner> entry : configuration.getSourceRunners().entrySet()) {
+ if (entry.getKey().equals(sourceName)) {
+ Source source = entry.getValue().getSource();
+ if (source instanceof EventProcessor) {
+ eventProcessor = (EventProcessor) source;
+ logger.debug("Found event processor {}", source.getName());
+ return;
+ }
+ }
+ }
+ logger.warn("No Source named {} found for republishing events.", sourceName);
}
@Override
@@ -83,6 +93,8 @@ public class NullSink extends AbstractSink implements Configurable, BatchSizeSup
Channel channel = getChannel();
Transaction transaction = channel.getTransaction();
Event event = null;
+ CounterGroup counterGroup = getCounterGroup();
+ long batchSize = getBatchSize();
long eventCounter = counterGroup.get("events.success");
try {
@@ -90,10 +102,13 @@ public class NullSink extends AbstractSink implements Configurable, BatchSizeSup
int i = 0;
for (i = 0; i < batchSize; i++) {
event = channel.take();
- if (++eventCounter % logEveryNEvents == 0) {
- logger.info("Null sink {} successful processed {} events.", getName(), eventCounter);
- }
- if (event == null) {
+ if (event != null) {
+ long id = Long.parseLong(new String(event.getBody()));
+ total += id;
+ event.getHeaders().put("Total", Long.toString(total));
+ eventProcessor.processEvent(event);
+ logger.info("Null sink {} successful processed event {}", getName(), id);
+ } else {
status = Status.BACKOFF;
break;
}
@@ -112,35 +127,4 @@ public class NullSink extends AbstractSink implements Configurable, BatchSizeSup
return status;
}
-
- @Override
- public void start() {
- logger.info("Starting {}...", this);
-
- counterGroup.setName(this.getName());
- super.start();
-
- logger.info("Null sink {} started.", getName());
- }
-
- @Override
- public void stop() {
- logger.info("Null sink {} stopping...", getName());
-
- super.stop();
-
- logger.info("Null sink {} stopped. Event metrics: {}",
- getName(), counterGroup);
- }
-
- @Override
- public String toString() {
- return "NullSink " + getName() + " { batchSize: " + batchSize + " }";
- }
-
- @Override
- public long getBatchSize() {
- return batchSize;
- }
-
}
diff --git a/flume-ng-node/src/test/java/org/apache/flume/source/EventProcessor.java b/flume-ng-node/src/test/java/org/apache/flume/source/EventProcessor.java
new file mode 100644
index 0000000..677065f
--- /dev/null
+++ b/flume-ng-node/src/test/java/org/apache/flume/source/EventProcessor.java
@@ -0,0 +1,33 @@
+/*
+ * 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.source;
+
+import org.apache.flume.Event;
+
+/**
+ * Interface indicating processEvent is implemented.
+ */
+public interface EventProcessor {
+ /**
+ * When implemented causes the event to be handled by the component.
+ * @param event The Flume event.
+ */
+ default void processEvent(Event event) {
+ }
+}
diff --git a/flume-ng-node/src/test/java/org/apache/flume/source/LocalSource.java b/flume-ng-node/src/test/java/org/apache/flume/source/LocalSource.java
new file mode 100644
index 0000000..18f609a
--- /dev/null
+++ b/flume-ng-node/src/test/java/org/apache/flume/source/LocalSource.java
@@ -0,0 +1,89 @@
+/*
+ * 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.source;
+
+import org.apache.flume.Context;
+import org.apache.flume.Event;
+import org.apache.flume.EventDrivenSource;
+import org.apache.flume.conf.Configurable;
+import org.apache.flume.instrumentation.SourceCounter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Accepts an event from a local component and publishes it to a channel.
+ */
+public class LocalSource extends AbstractSource implements Configurable, EventDrivenSource, EventProcessor {
+
+ private static final Logger logger = LoggerFactory.getLogger(LocalSource.class);
+
+ private SourceCounter sourceCounter;
+
+ /**
+ * Called by flume to start this source.
+ */
+ @Override
+ public void start() {
+ logger.info("Local source {} starting.", getName());
+ sourceCounter.start();
+ super.start();
+ }
+
+ /**
+ * Called by flume to stop this source.
+ */
+ @Override
+ public void stop() {
+ logger.info("Local source {} stopping.", getName());
+ sourceCounter.stop();
+ super.stop();
+ logger.info("Local source {} stopped. Metrics: {}", getName(), sourceCounter);
+ }
+
+ /**
+ * A message is passed in here. It is data that should be passed on.
+ *
+ * @param event The message.
+ */
+ public void processEvent(Event event) {
+ if (event == null) {
+ //Ignoring this. Not counting as an event received either.
+ return;
+ }
+
+ sourceCounter.incrementAppendReceivedCount();
+ sourceCounter.incrementEventReceivedCount();
+ logger.debug("pushing event to channel");
+ getChannelProcessor().processEvent(event);
+ sourceCounter.incrementAppendAcceptedCount();
+ sourceCounter.incrementEventAcceptedCount();
+ }
+
+ /**
+ * Called when flume starts up.
+ *
+ * @param context - Config values for this source from flume properties file.
+ */
+ @Override
+ public void configure(Context context) {
+ if (sourceCounter == null) {
+ sourceCounter = new SourceCounter(getName());
+ }
+ }
+}
diff --git a/flume-ng-node/src/test/resources/flume-conf-init.properties b/flume-ng-node/src/test/resources/flume-conf-init.properties
new file mode 100644
index 0000000..9ac5727
--- /dev/null
+++ b/flume-ng-node/src/test/resources/flume-conf-init.properties
@@ -0,0 +1,51 @@
+#
+# 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.
+#
+
+#
+# Flume Configuration
+# This file contains configuration for one Agent identified as host1.
+# This file also contains invalid configuration for few agents
+# host2, host3 etc.
+#
+
+host1.sources = source1 processedSource
+host1.channels = channel1 processedChannel
+host1.sinks = sink1
+
+# avroSource configuration
+host1.sources.source1.type = org.apache.flume.source.LocalSource
+host1.sources.source1.channels = channel1
+host1.sources.source1.totalEvents = 5
+host1.sources.source1.backoffSleepIncrement = 50
+
+host1.sources.processedSource.type = org.apache.flume.source.LocalSource
+host1.sources.processedSource.channels = processedChannel
+
+# memChannel1 configuration
+host1.channels.channel1.type = memory
+host1.channels.channel1.capacity = 10000
+
+host1.channels.processedChannel.type = memory
+host1.channels.processedChannel.capacity = 10000
+
+host1.sinks.sink1.type = org.apache.flume.sink.NullInitSink
+host1.sinks.sink1.batchSize = 1
+host1.sinks.sink1.targetSource = processedSource
+host1.sinks.sink1.channel = channel1
+
+
diff --git a/flume-ng-node/src/test/resources/log4j2.xml b/flume-ng-node/src/test/resources/log4j2.xml
index fbc44b4..732fef9 100644
--- a/flume-ng-node/src/test/resources/log4j2.xml
+++ b/flume-ng-node/src/test/resources/log4j2.xml
@@ -16,7 +16,7 @@
limitations under the License.
-->
-<Configuration status="OFF">
+<Configuration status="ERROR">
<Appenders>
<Console name="Console" target="SYSTEM_OUT">
<PatternLayout pattern="%d (%t) [%p - %l] %m%n" />
@@ -24,7 +24,7 @@
</Appenders>
<Loggers>
<Logger name="org.apache.flume" level="DEBUG"/>
- <Root level="INFO">
+ <Root level="DEBUG">
<AppenderRef ref="Console" />
</Root>
</Loggers>