This is an automated email from the ASF dual-hosted git repository.
rgoers pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/logging-flume-rpc.git
The following commit(s) were added to refs/heads/main by this push:
new c80723b Move embedded agent to this project
c80723b is described below
commit c80723b51ea7f8f6707adc530465add21e546466
Author: Ralph Goers <[email protected]>
AuthorDate: Sat May 30 12:50:45 2026 -0700
Move embedded agent to this project
---
.github/workflows/build.yml | 4 +-
flume-embedded-agent/pom.xml | 98 +++++++
.../apache/flume/agent/embedded/EmbeddedAgent.java | 259 +++++++++++++++++
.../agent/embedded/EmbeddedAgentConfiguration.java | 323 +++++++++++++++++++++
.../flume/agent/embedded/EmbeddedSource.java | 52 ++++
.../MaterializedConfigurationProvider.java | 42 +++
.../embedded/MemoryConfigurationProvider.java | 48 +++
.../apache/flume/agent/embedded/package-info.java | 24 ++
.../flume/agent/embedded/TestEmbeddedAgent.java | 230 +++++++++++++++
.../embedded/TestEmbeddedAgentConfiguration.java | 151 ++++++++++
.../embedded/TestEmbeddedAgentEmbeddedSource.java | 209 +++++++++++++
.../agent/embedded/TestEmbeddedAgentState.java | 125 ++++++++
flume-embedded-agent/src/test/resources/log4j2.xml | 31 ++
pom.xml | 1 +
14 files changed, 1595 insertions(+), 2 deletions(-)
diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml
index e786b72..284bc1a 100644
--- a/.github/workflows/build.yml
+++ b/.github/workflows/build.yml
@@ -52,11 +52,11 @@ jobs:
uses: actions/checkout@v6
# JDK 8 is needed for the build, and it is the primary bytecode target.
- - name: Setup JDK 8
+ - name: Setup JDK 17
uses: actions/setup-java@v5
with:
distribution: ${{ matrix.java-distribution }}
- java-version: 8
+ java-version: 17
cache: maven
- name: Build with Maven
diff --git a/flume-embedded-agent/pom.xml b/flume-embedded-agent/pom.xml
new file mode 100644
index 0000000..42c9ac3
--- /dev/null
+++ b/flume-embedded-agent/pom.xml
@@ -0,0 +1,98 @@
+<!--
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements. See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+-->
+<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>
+ <artifactId>flume-rpc</artifactId>
+ <groupId>org.apache.flume</groupId>
+ <version>2.0.0-SNAPSHOT</version>
+ <relativePath>../pom.xml</relativePath>
+ </parent>
+
+ <artifactId>flume-embedded-agent</artifactId>
+ <name>Flume Embedded Agent</name>
+ <description>Flume Embedded Agent: Stable public API for Embedding a Flume
2.x Agent</description>
+
+ <properties>
+ <!-- TODO fix pmd violations -->
+ <pmd.maxAllowedViolations>2</pmd.maxAllowedViolations>
+ <module.name>org.apache.flume.agent.embedded</module.name>
+ </properties>
+
+ <dependencies>
+
+ <dependency>
+ <groupId>org.apache.flume</groupId>
+ <artifactId>flume-ng-sdk</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flume</groupId>
+ <artifactId>flume-ng-configuration</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flume</groupId>
+ <artifactId>flume-ng-core</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flume</groupId>
+ <artifactId>flume-ng-node</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flume.flume-ng-channels</groupId>
+ <artifactId>flume-file-channel</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flume</groupId>
+ <artifactId>flume-rpc-avro</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-api</artifactId>
+ <!-- only used for tests -->
+ <optional>true</optional>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-1.2-api</artifactId>
+ <!-- only used for tests -->
+ <optional>true</optional>
+ </dependency>
+
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-core</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ </dependencies>
+</project>
diff --git
a/flume-embedded-agent/src/main/java/org/apache/flume/agent/embedded/EmbeddedAgent.java
b/flume-embedded-agent/src/main/java/org/apache/flume/agent/embedded/EmbeddedAgent.java
new file mode 100644
index 0000000..45c9e43
--- /dev/null
+++
b/flume-embedded-agent/src/main/java/org/apache/flume/agent/embedded/EmbeddedAgent.java
@@ -0,0 +1,259 @@
+/*
+ * 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.agent.embedded;
+
+import java.util.List;
+import java.util.Map;
+import java.util.TreeSet;
+
+import org.apache.flume.Channel;
+import org.apache.flume.exception.ChannelException;
+import org.apache.flume.Event;
+import org.apache.flume.EventDeliveryException;
+import org.apache.flume.FlumeException;
+import org.apache.flume.SinkRunner;
+import org.apache.flume.Source;
+import org.apache.flume.SourceRunner;
+import org.apache.flume.annotations.InterfaceAudience;
+import org.apache.flume.annotations.InterfaceStability;
+import org.apache.flume.conf.LogPrivacyUtil;
+import org.apache.flume.lifecycle.LifecycleAware;
+import org.apache.flume.lifecycle.LifecycleState;
+import org.apache.flume.lifecycle.LifecycleSupervisor;
+import org.apache.flume.lifecycle.LifecycleSupervisor.SupervisorPolicy;
+import org.apache.flume.node.MaterializedConfiguration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+
+/**
+ * EmbeddedAgent gives Flume users the ability to embed simple agents in
+ * applications. This Agent is mean to be much simpler than a traditional
+ * agent and as such it's more restrictive than what can be configured
+ * for a traditional agent. For specifics see the Flume Developer Guide.
+ */
[email protected]
[email protected]
+public class EmbeddedAgent {
+ private static final Logger LOGGER = LoggerFactory
+ .getLogger(EmbeddedAgent.class);
+
+ private final MaterializedConfigurationProvider configurationProvider;
+ private final String name;
+
+ private final LifecycleSupervisor supervisor;
+ private State state;
+ private SourceRunner sourceRunner;
+ private Channel channel;
+ private SinkRunner sinkRunner;
+ private EmbeddedSource embeddedSource;
+
+ @InterfaceAudience.Private
+ @InterfaceStability.Unstable
+ @VisibleForTesting
+ EmbeddedAgent(MaterializedConfigurationProvider configurationProvider,
+ String name) {
+ this.configurationProvider = configurationProvider;
+ this.name = name;
+ state = State.NEW;
+ supervisor = new LifecycleSupervisor();
+
+ }
+
+ public EmbeddedAgent(String name) {
+ this(new MaterializedConfigurationProvider(), name);
+ }
+
+ /**
+ * Configures the embedded agent. Can only be called after the object
+ * is created or after the stop() method is called.
+ *
+ * @param properties source, channel, and sink group configuration
+ * @throws FlumeException if a component is unable to be found or configured
+ * @throws IllegalStateException if called while the agent is started
+ */
+ public void configure(Map<String, String> properties)
+ throws FlumeException {
+ if (state == State.STARTED) {
+ throw new IllegalStateException("Cannot be configured while started");
+ }
+ doConfigure(properties);
+ state = State.STOPPED;
+ }
+
+ /**
+ * Started the agent. Can only be called after a successful call to
+ * configure().
+ *
+ * @throws FlumeException if a component cannot be started
+ * @throws IllegalStateException if the agent has not been configured or is
+ * already started
+ */
+ public void start()
+ throws FlumeException {
+ if (state == State.STARTED) {
+ throw new IllegalStateException("Cannot be started while started");
+ } else if (state == State.NEW) {
+ throw new IllegalStateException("Cannot be started before being " +
+ "configured");
+ }
+ // This check needs to be done before doStart(),
+ // as doStart() accesses sourceRunner.getSource()
+ Source source = Preconditions.checkNotNull(sourceRunner.getSource(),
+ "Source runner returned null source");
+ if (source instanceof EmbeddedSource) {
+ embeddedSource = (EmbeddedSource)source;
+ } else {
+ throw new IllegalStateException("Unknown source type: " +
source.getClass().getName());
+ }
+ doStart();
+ state = State.STARTED;
+ }
+
+ /**
+ * Stops the agent. Can only be called after a successful call to start().
+ * After a call to stop(), the agent can be re-configured with the
+ * configure() method or re-started with the start() method.
+ *
+ * @throws FlumeException if a component cannot be stopped
+ * @throws IllegalStateException if the agent is not started
+ */
+ public void stop()
+ throws FlumeException {
+ if (state != State.STARTED) {
+ throw new IllegalStateException("Cannot be stopped unless started");
+ }
+ supervisor.stop();
+ embeddedSource = null;
+ state = State.STOPPED;
+ }
+
+ private void doConfigure(Map<String, String> properties) {
+
+ properties = EmbeddedAgentConfiguration.configure(name, properties);
+
+ if (LOGGER.isDebugEnabled() && LogPrivacyUtil.allowLogPrintConfig()) {
+ LOGGER.debug("Agent configuration values");
+ for (String key : new TreeSet<String>(properties.keySet())) {
+ LOGGER.debug(key + " = " + properties.get(key));
+ }
+ }
+
+ MaterializedConfiguration conf = configurationProvider.get(name,
+ properties);
+ Map<String, SourceRunner> sources = conf.getSourceRunners();
+ if (sources.size() != 1) {
+ throw new FlumeException("Expected one source and got " +
+ sources.size());
+ }
+ Map<String, Channel> channels = conf.getChannels();
+ if (channels.size() != 1) {
+ throw new FlumeException("Expected one channel and got " +
+ channels.size());
+ }
+ Map<String, SinkRunner> sinks = conf.getSinkRunners();
+ if (sinks.size() != 1) {
+ throw new FlumeException("Expected one sink group and got " +
+ sinks.size());
+ }
+ this.sourceRunner = sources.values().iterator().next();
+ this.channel = channels.values().iterator().next();
+ this.sinkRunner = sinks.values().iterator().next();
+ }
+
+ /**
+ * Adds event to the channel owned by the agent. Note however, that the
+ * event is not copied and as such, the byte array and headers cannot
+ * be re-used by the caller.
+ * @param event
+ * @throws EventDeliveryException if unable to add event to channel
+ */
+ public void put(Event event) throws EventDeliveryException {
+ if (state != State.STARTED) {
+ throw new IllegalStateException("Cannot put events unless started");
+ }
+ try {
+ embeddedSource.put(event);
+ } catch (ChannelException ex) {
+ throw new EventDeliveryException("Embedded agent " + name +
+ ": Unable to process event: " + ex.getMessage(), ex);
+ }
+ }
+
+ /**
+ * Adds events to the channel owned by the agent. Note however, that the
+ * event is not copied and as such, the byte array and headers cannot
+ * be re-used by the caller.
+ * @param events
+ * @throws EventDeliveryException if unable to add event to channel
+ */
+ public void putAll(List<Event> events) throws EventDeliveryException {
+ if (state != State.STARTED) {
+ throw new IllegalStateException("Cannot put events unless started");
+ }
+ try {
+ embeddedSource.putAll(events);
+ } catch (ChannelException ex) {
+ throw new EventDeliveryException("Embedded agent " + name +
+ ": Unable to process event: " + ex.getMessage(), ex);
+ }
+ }
+
+ private void doStart() {
+ boolean error = true;
+ try {
+ channel.start();
+ sinkRunner.start();
+ sourceRunner.start();
+
+ supervisor.supervise(channel,
+ new SupervisorPolicy.AlwaysRestartPolicy(), LifecycleState.START);
+ supervisor.supervise(sinkRunner,
+ new SupervisorPolicy.AlwaysRestartPolicy(), LifecycleState.START);
+ supervisor.supervise(sourceRunner,
+ new SupervisorPolicy.AlwaysRestartPolicy(), LifecycleState.START);
+ error = false;
+ } finally {
+ if (error) {
+ stopLogError(sourceRunner);
+ stopLogError(channel);
+ stopLogError(sinkRunner);
+ supervisor.stop();
+ }
+ }
+ }
+
+ private void stopLogError(LifecycleAware lifeCycleAware) {
+ try {
+ if (LifecycleState.START.equals(lifeCycleAware.getLifecycleState())) {
+ lifeCycleAware.stop();
+ }
+ } catch (Exception e) {
+ LOGGER.warn("Exception while stopping " + lifeCycleAware, e);
+ }
+ }
+
+ private static enum State {
+ NEW(),
+ STOPPED(),
+ STARTED();
+ }
+}
diff --git
a/flume-embedded-agent/src/main/java/org/apache/flume/agent/embedded/EmbeddedAgentConfiguration.java
b/flume-embedded-agent/src/main/java/org/apache/flume/agent/embedded/EmbeddedAgentConfiguration.java
new file mode 100644
index 0000000..fe728f9
--- /dev/null
+++
b/flume-embedded-agent/src/main/java/org/apache/flume/agent/embedded/EmbeddedAgentConfiguration.java
@@ -0,0 +1,323 @@
+/*
+ * 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.agent.embedded;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.flume.FlumeException;
+import org.apache.flume.annotations.InterfaceAudience;
+import org.apache.flume.annotations.InterfaceStability;
+import org.apache.flume.conf.BasicConfigurationConstants;
+import org.apache.flume.conf.channel.ChannelType;
+import org.apache.flume.conf.sink.SinkProcessorType;
+import org.apache.flume.conf.sink.SinkType;
+
+import com.google.common.base.Joiner;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Maps;
+
+/**
+ * Stores publicly accessible configuration constants and private
+ * configuration constants and methods.
+ */
[email protected]
[email protected]
+public class EmbeddedAgentConfiguration {
+ public static final String SEPERATOR = ".";
+ private static final Joiner JOINER = Joiner.on(SEPERATOR);
+ private static final String TYPE = "type";
+
+ /**
+ * Prefix for source properties
+ */
+ public static final String SOURCE = "source";
+ /**
+ * Prefix for channel properties
+ */
+ public static final String CHANNEL = "channel";
+ /**
+ * Prefix for sink processor properties
+ */
+ public static final String SINK_PROCESSOR = "processor";
+ /**
+ * Space delimited list of sink names: e.g. sink1 sink2 sink3
+ */
+ public static final String SINKS = "sinks";
+
+ public static final String SINKS_PREFIX = join(SINKS, "");
+ /**
+ * Source type, choices are `embedded'
+ */
+ public static final String SOURCE_TYPE = join(SOURCE, TYPE);
+ /**
+ * Prefix for passing configuration parameters to the source
+ */
+ public static final String SOURCE_PREFIX = join(SOURCE, "");
+ /**
+ * Channel type, choices are `memory' or `file'
+ */
+ public static final String CHANNEL_TYPE = join(CHANNEL, TYPE);
+ /**
+ * Prefix for passing configuration parameters to the channel
+ */
+ public static final String CHANNEL_PREFIX = join(CHANNEL, "");
+ /**
+ * Sink processor type, choices are `default', `failover' or `load_balance'
+ */
+ public static final String SINK_PROCESSOR_TYPE = join(SINK_PROCESSOR, TYPE);
+ /**
+ * Prefix for passing configuration parameters to the sink processor
+ */
+ public static final String SINK_PROCESSOR_PREFIX = join(SINK_PROCESSOR, "");
+ /**
+ * Embedded source which provides simple in-memory transfer to channel.
+ * Use this source via the put,putAll methods on the EmbeddedAgent. This
+ * is the only supported source to use for Embedded Agents.
+ */
+ public static final String SOURCE_TYPE_EMBEDDED =
EmbeddedSource.class.getName();
+ private static final String SOURCE_TYPE_EMBEDDED_ALIAS = "EMBEDDED";
+ /**
+ * Memory channel which stores events in heap. See Flume User Guide for
+ * configuration information. This is the recommended channel to use for
+ * Embedded Agents.
+ */
+ public static final String CHANNEL_TYPE_MEMORY = ChannelType.MEMORY.name();
+
+ /**
+ * Spillable Memory channel which stores events in heap. See Flume User Guide
for
+ * configuration information. This is the recommended channel to use for
+ * Embedded Agents.
+ */
+ public static final String CHANNEL_TYPE_SPILLABLEMEMORY =
ChannelType.SPILLABLEMEMORY.name();
+
+ /**
+ * File based channel which stores events in on local disk. See Flume User
+ * Guide for configuration information.
+ */
+ public static final String CHANNEL_TYPE_FILE = ChannelType.FILE.name();
+
+ /**
+ * Avro sink which can send events to a downstream avro source. This is the
+ * only supported sink for Embedded Agents.
+ */
+ public static final String SINK_TYPE_AVRO = SinkType.AVRO.name();
+
+ /**
+ * Default sink processors which may be used when there is only a single
sink.
+ */
+ public static final String SINK_PROCESSOR_TYPE_DEFAULT =
SinkProcessorType.DEFAULT.name();
+ /**
+ * Failover sink processor. See Flume User Guide for configuration
+ * information.
+ */
+ public static final String SINK_PROCESSOR_TYPE_FAILOVER =
SinkProcessorType.FAILOVER.name();
+ /**
+ * Load balancing sink processor. See Flume User Guide for configuration
+ * information.
+ */
+ public static final String SINK_PROCESSOR_TYPE_LOAD_BALANCE =
+ SinkProcessorType.LOAD_BALANCE.name();
+
+ private static final String[] ALLOWED_SOURCES = {
+ SOURCE_TYPE_EMBEDDED_ALIAS,
+ SOURCE_TYPE_EMBEDDED,
+ };
+
+ private static final String[] ALLOWED_CHANNELS = {
+ CHANNEL_TYPE_MEMORY,
+ CHANNEL_TYPE_FILE
+ };
+
+ private static final String[] ALLOWED_SINKS = {
+ SINK_TYPE_AVRO
+ };
+
+ private static final String[] ALLOWED_SINK_PROCESSORS = {
+ SINK_PROCESSOR_TYPE_DEFAULT,
+ SINK_PROCESSOR_TYPE_FAILOVER,
+ SINK_PROCESSOR_TYPE_LOAD_BALANCE
+ };
+
+ private static final ImmutableList<String> DISALLOWED_SINK_NAMES =
+ ImmutableList.of("source", "channel", "processor");
+
+ private static void validate(String name,
+ Map<String, String> properties) throws FlumeException {
+
+ if (properties.containsKey(SOURCE_TYPE)) {
+ checkAllowed(ALLOWED_SOURCES, properties.get(SOURCE_TYPE));
+ }
+ checkRequired(properties, CHANNEL_TYPE);
+ checkAllowed(ALLOWED_CHANNELS, properties.get(CHANNEL_TYPE));
+ checkRequired(properties, SINKS);
+ String sinkNames = properties.get(SINKS);
+ for (String sink : sinkNames.split("\\s+")) {
+ if (DISALLOWED_SINK_NAMES.contains(sink.toLowerCase(Locale.ENGLISH))) {
+ throw new FlumeException("Sink name " + sink + " is one of the" +
+ " disallowed sink names: " + DISALLOWED_SINK_NAMES);
+ }
+ String key = join(sink, TYPE);
+ checkRequired(properties, key);
+ checkAllowed(ALLOWED_SINKS, properties.get(key));
+ }
+ checkRequired(properties, SINK_PROCESSOR_TYPE);
+ checkAllowed(ALLOWED_SINK_PROCESSORS, properties.get(SINK_PROCESSOR_TYPE));
+ }
+ /**
+ * Folds embedded configuration structure into an agent configuration.
+ * Should only be called after validate returns without error.
+ *
+ * @param name - agent name
+ * @param properties - embedded agent configuration
+ * @return configuration applicable to a flume agent
+ */
+ @InterfaceAudience.Private
+ @InterfaceStability.Unstable
+ static Map<String, String> configure(String name,
+ Map<String, String> properties) throws FlumeException {
+ validate(name, properties);
+ // we are going to modify the properties as we parse the config
+ properties = new HashMap<String, String>(properties);
+
+ if (!properties.containsKey(SOURCE_TYPE) ||
+
SOURCE_TYPE_EMBEDDED_ALIAS.equalsIgnoreCase(properties.get(SOURCE_TYPE))) {
+ properties.put(SOURCE_TYPE, SOURCE_TYPE_EMBEDDED);
+ }
+ String sinkNames = properties.remove(SINKS);
+
+ String strippedName = name.replaceAll("\\s+","");
+ String sourceName = "source-" + strippedName;
+ String channelName = "channel-" + strippedName;
+ String sinkGroupName = "sink-group-" + strippedName;
+
+ /*
+ * Now we are going to process the user supplied configuration
+ * and generate an agent configuration. This is only to supply
+ * a simpler client api than passing in an entire agent configuration.
+ */
+ // user supplied config -> agent configuration
+ Map<String, String> result = Maps.newHashMap();
+
+ /*
+ * First we are going to setup all the root level pointers. I.E
+ * point the agent at the components, sink group at sinks, and
+ * source at the channel.
+ */
+ // point agent at source
+ result.put(join(name, BasicConfigurationConstants.CONFIG_SOURCES),
+ sourceName);
+ // point agent at channel
+ result.put(join(name, BasicConfigurationConstants.CONFIG_CHANNELS),
+ channelName);
+ // point agent at sinks
+ result.put(join(name, BasicConfigurationConstants.CONFIG_SINKS),
+ sinkNames);
+ // points the agent at the sinkgroup
+ result.put(join(name, BasicConfigurationConstants.CONFIG_SINKGROUPS),
+ sinkGroupName);
+ // points the sinkgroup at the sinks
+ result.put(join(name, BasicConfigurationConstants.CONFIG_SINKGROUPS,
+ sinkGroupName, SINKS), sinkNames);
+ // points the source at the channel
+ result.put(join(name,
+ BasicConfigurationConstants.CONFIG_SOURCES, sourceName,
+ BasicConfigurationConstants.CONFIG_CHANNELS), channelName);
+
+ // Properties will be modified during iteration so we need a
+ // copy of the keys.
+ Set<String> userProvidedKeys = new HashSet<String>(properties.keySet());
+
+ /*
+ * Second process the sink configuration and point the sinks
+ * at the channel.
+ */
+ for (String sink : sinkNames.split("\\s+")) {
+ for (String key : userProvidedKeys) {
+ String value = properties.get(key);
+ if (key.startsWith(sink + SEPERATOR)) {
+ properties.remove(key);
+ result.put(join(name,
+ BasicConfigurationConstants.CONFIG_SINKS, key), value);
+ }
+ }
+ // point the sink at the channel
+ result.put(join(name,
+ BasicConfigurationConstants.CONFIG_SINKS, sink,
+ BasicConfigurationConstants.CONFIG_CHANNEL), channelName);
+ }
+ /*
+ * Third, process all remaining configuration items, prefixing them
+ * correctly and then passing them on to the agent.
+ */
+ userProvidedKeys = new HashSet<String>(properties.keySet());
+ for (String key : userProvidedKeys) {
+ String value = properties.get(key);
+ if (key.startsWith(SOURCE_PREFIX)) {
+ // users use `source' but agent needs the actual source name
+ key = key.replaceFirst(SOURCE, sourceName);
+ result.put(join(name,
+ BasicConfigurationConstants.CONFIG_SOURCES, key), value);
+ } else if (key.startsWith(CHANNEL_PREFIX)) {
+ // users use `channel' but agent needs the actual channel name
+ key = key.replaceFirst(CHANNEL, channelName);
+ result.put(join(name,
+ BasicConfigurationConstants.CONFIG_CHANNELS, key), value);
+ } else if (key.startsWith(SINK_PROCESSOR_PREFIX)) {
+ // agent.sinkgroups.sinkgroup.processor.*
+ result.put(join(name, BasicConfigurationConstants.CONFIG_SINKGROUPS,
+ sinkGroupName, key), value);
+ } else {
+ // XXX should we simply ignore this?
+ throw new FlumeException("Unknown configuration " + key);
+ }
+ }
+ return result;
+ }
+ private static void checkAllowed(String[] allowedTypes, String type) {
+ boolean isAllowed = false;
+ type = type.trim();
+ for (String allowedType : allowedTypes) {
+ if (allowedType.equalsIgnoreCase(type)) {
+ isAllowed = true;
+ break;
+ }
+ }
+ if (!isAllowed) {
+ throw new FlumeException("Component type of " + type + " is not in " +
+ "allowed types of " + Arrays.toString(allowedTypes));
+ }
+ }
+ private static void checkRequired(Map<String, String> properties, String
name) {
+ if (!properties.containsKey(name)) {
+ throw new FlumeException("Required parameter not found " + name);
+ }
+ }
+
+ private static String join(String... parts) {
+ return JOINER.join(parts);
+ }
+
+ private EmbeddedAgentConfiguration() {
+ }
+}
\ No newline at end of file
diff --git
a/flume-embedded-agent/src/main/java/org/apache/flume/agent/embedded/EmbeddedSource.java
b/flume-embedded-agent/src/main/java/org/apache/flume/agent/embedded/EmbeddedSource.java
new file mode 100644
index 0000000..375e27b
--- /dev/null
+++
b/flume-embedded-agent/src/main/java/org/apache/flume/agent/embedded/EmbeddedSource.java
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.flume.agent.embedded;
+
+import java.util.List;
+
+import org.apache.flume.exception.ChannelException;
+import org.apache.flume.Context;
+import org.apache.flume.Event;
+import org.apache.flume.EventDrivenSource;
+import org.apache.flume.annotations.InterfaceAudience;
+import org.apache.flume.annotations.InterfaceStability;
+import org.apache.flume.conf.Configurable;
+import org.apache.flume.source.AbstractSource;
+
+/**
+ * Simple source used to allow direct access to the channel for the Embedded
+ * Agent.
+ */
[email protected]
[email protected]
+public class EmbeddedSource extends AbstractSource implements
EventDrivenSource, Configurable {
+
+ @Override
+ public void configure(Context context) {
+ }
+
+ public void put(Event event) throws ChannelException {
+ getChannelProcessor().processEvent(event);
+ }
+
+ public void putAll(List<Event> events) throws ChannelException {
+ getChannelProcessor().processEventBatch(events);
+ }
+
+}
diff --git
a/flume-embedded-agent/src/main/java/org/apache/flume/agent/embedded/MaterializedConfigurationProvider.java
b/flume-embedded-agent/src/main/java/org/apache/flume/agent/embedded/MaterializedConfigurationProvider.java
new file mode 100644
index 0000000..9d93138
--- /dev/null
+++
b/flume-embedded-agent/src/main/java/org/apache/flume/agent/embedded/MaterializedConfigurationProvider.java
@@ -0,0 +1,42 @@
+/*
+ * 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.agent.embedded;
+
+import java.util.Map;
+
+import org.apache.flume.annotations.InterfaceAudience;
+import org.apache.flume.annotations.InterfaceStability;
+import org.apache.flume.node.MaterializedConfiguration;
+
+/**
+ * Provides {@see MaterializedConfiguration} for a given agent and set of
+ * properties. This class exists simply to make more easily testable. That is
+ * it allows us to mock the actual Source, Sink, and Channel components
+ * as opposed to instantiation of real components.
+ */
[email protected]
[email protected]
+class MaterializedConfigurationProvider {
+
+ MaterializedConfiguration get(String name, Map<String, String> properties) {
+ MemoryConfigurationProvider confProvider =
+ new MemoryConfigurationProvider(name, properties);
+ return confProvider.getConfiguration();
+ }
+}
diff --git
a/flume-embedded-agent/src/main/java/org/apache/flume/agent/embedded/MemoryConfigurationProvider.java
b/flume-embedded-agent/src/main/java/org/apache/flume/agent/embedded/MemoryConfigurationProvider.java
new file mode 100644
index 0000000..0377501
--- /dev/null
+++
b/flume-embedded-agent/src/main/java/org/apache/flume/agent/embedded/MemoryConfigurationProvider.java
@@ -0,0 +1,48 @@
+/*
+ * 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.agent.embedded;
+
+import java.util.Map;
+
+import org.apache.flume.annotations.InterfaceAudience;
+import org.apache.flume.annotations.InterfaceStability;
+import org.apache.flume.conf.FlumeConfiguration;
+import org.apache.flume.node.AbstractConfigurationProvider;
+
+/**
+ * MemoryConfigurationProvider is the simplest possible
+ * AbstractConfigurationProvider simply turning a give properties file and
+ * agent name into a FlumeConfiguration object.
+ */
[email protected]
[email protected]
+class MemoryConfigurationProvider extends AbstractConfigurationProvider {
+ private final Map<String, String> properties;
+
+ MemoryConfigurationProvider(String name, Map<String, String> properties) {
+ super(name);
+ this.properties = properties;
+ }
+
+ @Override
+ protected FlumeConfiguration getFlumeConfiguration() {
+ return new FlumeConfiguration(properties);
+ }
+
+}
diff --git
a/flume-embedded-agent/src/main/java/org/apache/flume/agent/embedded/package-info.java
b/flume-embedded-agent/src/main/java/org/apache/flume/agent/embedded/package-info.java
new file mode 100644
index 0000000..919a630
--- /dev/null
+++
b/flume-embedded-agent/src/main/java/org/apache/flume/agent/embedded/package-info.java
@@ -0,0 +1,24 @@
+/*
+ * 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.
+ */
+/**
+ * This package provides Flume users the ability to embed simple agents
+ * in applications. For specific and up to date information, please see
+ * the Flume Developer Guide.
+ */
+package org.apache.flume.agent.embedded;
diff --git
a/flume-embedded-agent/src/test/java/org/apache/flume/agent/embedded/TestEmbeddedAgent.java
b/flume-embedded-agent/src/test/java/org/apache/flume/agent/embedded/TestEmbeddedAgent.java
new file mode 100644
index 0000000..68a3a07
--- /dev/null
+++
b/flume-embedded-agent/src/test/java/org/apache/flume/agent/embedded/TestEmbeddedAgent.java
@@ -0,0 +1,230 @@
+/*
+ * 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.agent.embedded;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.ServerSocket;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Queue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.avro.ipc.netty.NettyServer;
+import org.apache.avro.ipc.Responder;
+import org.apache.avro.ipc.specific.SpecificResponder;
+import org.apache.flume.Event;
+import org.apache.flume.event.EventBuilder;
+import org.apache.flume.source.avro.AvroFlumeEvent;
+import org.apache.flume.source.avro.AvroSourceProtocol;
+import org.apache.flume.source.avro.Status;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.base.Charsets;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+
+public class TestEmbeddedAgent {
+ private static final Logger LOGGER = LoggerFactory
+ .getLogger(TestEmbeddedAgent.class);
+ private static final String HOSTNAME = "localhost";
+ private static AtomicInteger serialNumber = new AtomicInteger(0);
+ private EmbeddedAgent agent;
+ private Map<String, String> properties;
+ private EventCollector eventCollector;
+ private NettyServer nettyServer;
+ private Map<String, String> headers;
+ private byte[] body;
+
+ @Before
+ public void setUp() throws Exception {
+ headers = Maps.newHashMap();
+ headers.put("key1", "value1");
+ body = "body".getBytes(Charsets.UTF_8);
+
+ int port = findFreePort();
+ eventCollector = new EventCollector();
+ Responder responder = new SpecificResponder(AvroSourceProtocol.class,
+ eventCollector);
+ nettyServer = new NettyServer(responder,
+ new InetSocketAddress(HOSTNAME, port));
+ nettyServer.start();
+
+ // give the server a second to start
+ Thread.sleep(1000L);
+
+ properties = Maps.newHashMap();
+ properties.put("channel.type", "memory");
+ properties.put("channel.capacity", "200");
+ properties.put("sinks", "sink1 sink2");
+ properties.put("sink1.type", "avro");
+ properties.put("sink2.type", "avro");
+ properties.put("sink1.hostname", HOSTNAME);
+ properties.put("sink1.port", String.valueOf(port));
+ properties.put("sink2.hostname", HOSTNAME);
+ properties.put("sink2.port", String.valueOf(port));
+ properties.put("processor.type", "load_balance");
+
+ agent = new EmbeddedAgent("test-" + serialNumber.incrementAndGet());
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ if (agent != null) {
+ try {
+ agent.stop();
+ } catch (Exception e) {
+ LOGGER.debug("Error shutting down agent", e);
+ }
+ }
+ if (nettyServer != null) {
+ try {
+ nettyServer.close();
+ } catch (Exception e) {
+ LOGGER.debug("Error shutting down server", e);
+ }
+ }
+ }
+
+ @Test(timeout = 30000L)
+ public void testPut() throws Exception {
+ agent.configure(properties);
+ agent.start();
+ agent.put(EventBuilder.withBody(body, headers));
+
+ Event event;
+ while ((event = eventCollector.poll()) == null) {
+ Thread.sleep(500L);
+ }
+ Assert.assertNotNull(event);
+ Assert.assertArrayEquals(body, event.getBody());
+ Assert.assertEquals(headers, event.getHeaders());
+ }
+
+ @Test(timeout = 30000L)
+ public void testPutAll() throws Exception {
+ List<Event> events = Lists.newArrayList();
+ events.add(EventBuilder.withBody(body, headers));
+ agent.configure(properties);
+ agent.start();
+ agent.putAll(events);
+
+ Event event;
+ while ((event = eventCollector.poll()) == null) {
+ Thread.sleep(500L);
+ }
+ Assert.assertNotNull(event);
+ Assert.assertArrayEquals(body, event.getBody());
+ Assert.assertEquals(headers, event.getHeaders());
+ }
+
+ @Test(timeout = 30000L)
+ public void testPutWithInterceptors() throws Exception {
+ properties.put("source.interceptors", "i1");
+ properties.put("source.interceptors.i1.type", "static");
+ properties.put("source.interceptors.i1.key", "key2");
+ properties.put("source.interceptors.i1.value", "value2");
+
+ agent.configure(properties);
+ agent.start();
+ agent.put(EventBuilder.withBody(body, headers));
+
+ Event event;
+ while ((event = eventCollector.poll()) == null) {
+ Thread.sleep(500L);
+ }
+ Assert.assertNotNull(event);
+ Assert.assertArrayEquals(body, event.getBody());
+ Map<String, String> newHeaders = new HashMap<String, String>(headers);
+ newHeaders.put("key2", "value2");
+ Assert.assertEquals(newHeaders, event.getHeaders());
+ }
+
+
+ @Test(timeout = 30000L)
+ public void testEmbeddedAgentName() throws Exception {
+ EmbeddedAgent embedAgent = new EmbeddedAgent("test 1 2" +
serialNumber.incrementAndGet());
+ List<Event> events = Lists.newArrayList();
+ events.add(EventBuilder.withBody(body, headers));
+ embedAgent.configure(properties);
+ embedAgent.start();
+ embedAgent.putAll(events);
+
+ Event event;
+ while ((event = eventCollector.poll()) == null) {
+ Thread.sleep(500L);
+ }
+ Assert.assertNotNull(event);
+ Assert.assertArrayEquals(body, event.getBody());
+ Assert.assertEquals(headers, event.getHeaders());
+ if (embedAgent != null) {
+ try {
+ embedAgent.stop();
+ } catch (Exception e) {
+ LOGGER.debug("Error shutting down agent", e);
+ }
+ }
+ }
+
+ static class EventCollector implements AvroSourceProtocol {
+ private final Queue<AvroFlumeEvent> eventQueue =
+ new LinkedBlockingQueue<AvroFlumeEvent>();
+
+ public Event poll() {
+ AvroFlumeEvent avroEvent = eventQueue.poll();
+ if (avroEvent != null) {
+ return EventBuilder.withBody(avroEvent.getBody().array(),
+ toStringMap(avroEvent.getHeaders()));
+ }
+ return null;
+ }
+ @Override
+ public Status append(AvroFlumeEvent event) {
+ eventQueue.add(event);
+ return Status.OK;
+ }
+ @Override
+ public Status appendBatch(List<AvroFlumeEvent> events) {
+ Preconditions.checkState(eventQueue.addAll(events));
+ return Status.OK;
+ }
+ }
+
+ private static Map<String, String> toStringMap(Map<CharSequence,
CharSequence> charSeqMap) {
+ Map<String, String> stringMap = new HashMap<String, String>();
+ for (Map.Entry<CharSequence, CharSequence> entry : charSeqMap.entrySet()) {
+ stringMap.put(entry.getKey().toString(), entry.getValue().toString());
+ }
+ return stringMap;
+ }
+
+ private static int findFreePort() throws IOException {
+ try (ServerSocket socket = new ServerSocket(0)) {
+ return socket.getLocalPort();
+ }
+ }
+}
diff --git
a/flume-embedded-agent/src/test/java/org/apache/flume/agent/embedded/TestEmbeddedAgentConfiguration.java
b/flume-embedded-agent/src/test/java/org/apache/flume/agent/embedded/TestEmbeddedAgentConfiguration.java
new file mode 100644
index 0000000..ed26294
--- /dev/null
+++
b/flume-embedded-agent/src/test/java/org/apache/flume/agent/embedded/TestEmbeddedAgentConfiguration.java
@@ -0,0 +1,151 @@
+/*
+ * 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.agent.embedded;
+
+import java.util.Map;
+
+import junit.framework.Assert;
+
+import org.apache.flume.FlumeException;
+import org.junit.Before;
+import org.junit.Test;
+
+import com.google.common.collect.Maps;
+
+public class TestEmbeddedAgentConfiguration {
+ private Map<String, String> properties;
+
+ @Before
+ public void setUp() throws Exception {
+ properties = Maps.newHashMap();
+ properties.put("source.type",
EmbeddedAgentConfiguration.SOURCE_TYPE_EMBEDDED);
+ properties.put("channel.type", "memory");
+ properties.put("channel.capacity", "200");
+ properties.put("sinks", "sink1 sink2");
+ properties.put("sink1.type", "avro");
+ properties.put("sink2.type", "avro");
+ properties.put("sink1.hostname", "sink1.host");
+ properties.put("sink1.port", "2");
+ properties.put("sink2.hostname", "sink2.host");
+ properties.put("sink2.port", "2");
+ properties.put("processor.type", "load_balance");
+ properties.put("source.interceptors", "i1");
+ properties.put("source.interceptors.i1.type", "timestamp");
+ }
+
+ @Test
+ public void testFullSourceType() throws Exception {
+ doTestExcepted(EmbeddedAgentConfiguration.configure("test1", properties));
+ }
+
+ @Test
+ public void testMissingSourceType() throws Exception {
+ Assert.assertNotNull(properties.remove("source.type"));
+ doTestExcepted(EmbeddedAgentConfiguration.configure("test1", properties));
+ }
+
+ @Test
+ public void testShortSourceType() throws Exception {
+ properties.put("source.type", "EMBEDDED");
+ doTestExcepted(EmbeddedAgentConfiguration.configure("test1", properties));
+ }
+
+ public void doTestExcepted(Map<String, String> actual) throws Exception {
+ Map<String, String> expected = Maps.newHashMap();
+ expected.put("test1.channels", "channel-test1");
+ expected.put("test1.channels.channel-test1.capacity", "200");
+ expected.put("test1.channels.channel-test1.type", "memory");
+ expected.put("test1.sinkgroups", "sink-group-test1");
+ expected.put("test1.sinkgroups.sink-group-test1.processor.type",
"load_balance");
+ expected.put("test1.sinkgroups.sink-group-test1.sinks", "sink1 sink2");
+ expected.put("test1.sinks", "sink1 sink2");
+ expected.put("test1.sinks.sink1.channel", "channel-test1");
+ expected.put("test1.sinks.sink1.hostname", "sink1.host");
+ expected.put("test1.sinks.sink1.port", "2");
+ expected.put("test1.sinks.sink1.type", "avro");
+ expected.put("test1.sinks.sink2.channel", "channel-test1");
+ expected.put("test1.sinks.sink2.hostname", "sink2.host");
+ expected.put("test1.sinks.sink2.port", "2");
+ expected.put("test1.sinks.sink2.type", "avro");
+ expected.put("test1.sources", "source-test1");
+ expected.put("test1.sources.source-test1.channels", "channel-test1");
+ expected.put("test1.sources.source-test1.type",
+ EmbeddedAgentConfiguration.SOURCE_TYPE_EMBEDDED);
+ expected.put("test1.sources.source-test1.interceptors", "i1");
+ expected.put("test1.sources.source-test1.interceptors.i1.type",
"timestamp");
+ Assert.assertEquals(expected, actual);
+ }
+
+ @Test(expected = FlumeException.class)
+ public void testBadSource() throws Exception {
+ properties.put("source.type", "exec");
+ EmbeddedAgentConfiguration.configure("test1", properties);
+ }
+ @Test(expected = FlumeException.class)
+ public void testBadChannel() throws Exception {
+ properties.put("channel.type", "jdbc");
+ EmbeddedAgentConfiguration.configure("test1", properties);
+ }
+ @Test(expected = FlumeException.class)
+ public void testBadSink() throws Exception {
+ properties.put("sink1.type", "hbase");
+ EmbeddedAgentConfiguration.configure("test1", properties);
+ }
+ @Test(expected = FlumeException.class)
+ public void testBadSinkProcessor() throws Exception {
+ properties.put("processor.type", "bad");
+ EmbeddedAgentConfiguration.configure("test1", properties);
+ }
+
+ @Test(expected = FlumeException.class)
+ public void testNoChannel() throws Exception {
+ properties.remove("channel.type");
+ EmbeddedAgentConfiguration.configure("test1", properties);
+ }
+ @Test(expected = FlumeException.class)
+ public void testNoSink() throws Exception {
+ properties.remove("sink2.type");
+ EmbeddedAgentConfiguration.configure("test1", properties);
+ }
+ @Test(expected = FlumeException.class)
+ public void testNoSinkProcessor() throws Exception {
+ properties.remove("processor.type");
+ EmbeddedAgentConfiguration.configure("test1", properties);
+ }
+ @Test(expected = FlumeException.class)
+ public void testBadKey() throws Exception {
+ properties.put("bad.key.name", "bad");
+ EmbeddedAgentConfiguration.configure("test1", properties);
+ }
+ @Test(expected = FlumeException.class)
+ public void testSinkNamedLikeSource() throws Exception {
+ properties.put("sinks", "source");
+ EmbeddedAgentConfiguration.configure("test1", properties);
+ }
+ @Test(expected = FlumeException.class)
+ public void testSinkNamedLikeChannel() throws Exception {
+ properties.put("sinks", "channel");
+ EmbeddedAgentConfiguration.configure("test1", properties);
+ }
+ @Test(expected = FlumeException.class)
+ public void testSinkNamedLikeProcessor() throws Exception {
+ properties.put("sinks", "processor");
+ EmbeddedAgentConfiguration.configure("test1", properties);
+ }
+}
\ No newline at end of file
diff --git
a/flume-embedded-agent/src/test/java/org/apache/flume/agent/embedded/TestEmbeddedAgentEmbeddedSource.java
b/flume-embedded-agent/src/test/java/org/apache/flume/agent/embedded/TestEmbeddedAgentEmbeddedSource.java
new file mode 100644
index 0000000..ae59444
--- /dev/null
+++
b/flume-embedded-agent/src/test/java/org/apache/flume/agent/embedded/TestEmbeddedAgentEmbeddedSource.java
@@ -0,0 +1,209 @@
+/*
+ * 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.agent.embedded;
+
+import static org.mockito.Mockito.*;
+
+import java.util.List;
+import java.util.Map;
+
+import junit.framework.Assert;
+
+import org.apache.flume.Channel;
+import org.apache.flume.Event;
+import org.apache.flume.EventDeliveryException;
+import org.apache.flume.SinkRunner;
+import org.apache.flume.SourceRunner;
+import org.apache.flume.event.SimpleEvent;
+import org.apache.flume.lifecycle.LifecycleState;
+import org.apache.flume.node.MaterializedConfiguration;
+import org.junit.Before;
+import org.junit.Test;
+
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+
+public class TestEmbeddedAgentEmbeddedSource {
+
+ private EmbeddedAgent agent;
+ private Map<String, String> properties;
+
+ private MaterializedConfiguration config;
+ private EmbeddedSource source;
+ private SourceRunner sourceRunner;
+ private Channel channel;
+ private SinkRunner sinkRunner;
+
+ @Before
+ public void setUp() throws Exception {
+
+ properties = Maps.newHashMap();
+ properties.put("source.type",
EmbeddedAgentConfiguration.SOURCE_TYPE_EMBEDDED);
+ properties.put("channel.type", "memory");
+ properties.put("sinks", "sink1 sink2");
+ properties.put("sink1.type", "avro");
+ properties.put("sink2.type", "avro");
+ properties.put("processor.type", "load_balance");
+
+ sourceRunner = mock(SourceRunner.class);
+ channel = mock(Channel.class);
+ sinkRunner = mock(SinkRunner.class);
+
+ source = mock(EmbeddedSource.class);
+ when(sourceRunner.getSource()).thenReturn(source);
+
+ when(sourceRunner.getLifecycleState()).thenReturn(LifecycleState.START);
+ when(channel.getLifecycleState()).thenReturn(LifecycleState.START);
+ when(sinkRunner.getLifecycleState()).thenReturn(LifecycleState.START);
+
+ config = new MaterializedConfiguration() {
+ @Override
+ public Map<String, SourceRunner> getSourceRunners() {
+ Map<String, SourceRunner> result = Maps.newHashMap();
+ result.put("source", sourceRunner);
+ return ImmutableMap.copyOf(result);
+ }
+
+ @Override
+ public Map<String, SinkRunner> getSinkRunners() {
+ Map<String, SinkRunner> result = Maps.newHashMap();
+ result.put("sink", sinkRunner);
+ return ImmutableMap.copyOf(result);
+ }
+
+ @Override
+ public Map<String, Channel> getChannels() {
+ Map<String, Channel> result = Maps.newHashMap();
+ result.put("channel", channel);
+ return ImmutableMap.copyOf(result);
+ }
+
+ @Override
+ public void addSourceRunner(String name, SourceRunner sourceRunner) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void addSinkRunner(String name, SinkRunner sinkRunner) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void addChannel(String name, Channel channel) {
+ throw new UnsupportedOperationException();
+ }
+ };
+ agent = new EmbeddedAgent(new MaterializedConfigurationProvider() {
+ public MaterializedConfiguration get(String name, Map<String, String>
properties) {
+ return config;
+ }
+ }, "dummy");
+ }
+
+ @Test
+ public void testStart() {
+ agent.configure(properties);
+ agent.start();
+ verify(sourceRunner, times(1)).start();
+ verify(channel, times(1)).start();
+ verify(sinkRunner, times(1)).start();
+ }
+
+ @Test
+ public void testStop() {
+ agent.configure(properties);
+ agent.start();
+ agent.stop();
+ verify(sourceRunner, times(1)).stop();
+ verify(channel, times(1)).stop();
+ verify(sinkRunner, times(1)).stop();
+ }
+
+ @Test
+ public void testStartSourceThrowsException() {
+ doThrow(new LocalRuntimeException()).when(sourceRunner).start();
+ startExpectingLocalRuntimeException();
+ }
+
+ @Test
+ public void testStartChannelThrowsException() {
+ doThrow(new LocalRuntimeException()).when(channel).start();
+ startExpectingLocalRuntimeException();
+ }
+
+ @Test
+ public void testStartSinkThrowsException() {
+ doThrow(new LocalRuntimeException()).when(sinkRunner).start();
+ startExpectingLocalRuntimeException();
+ }
+
+ private void startExpectingLocalRuntimeException() {
+ agent.configure(properties);
+ try {
+ agent.start();
+ Assert.fail();
+ } catch (LocalRuntimeException e) {
+ // expected
+ }
+ verify(sourceRunner, times(1)).stop();
+ verify(channel, times(1)).stop();
+ verify(sinkRunner, times(1)).stop();
+ }
+
+ private static class LocalRuntimeException extends RuntimeException {
+ private static final long serialVersionUID = 116546244849853151L;
+ }
+
+ @Test
+ public void testPut() throws EventDeliveryException {
+ Event event = new SimpleEvent();
+ agent.configure(properties);
+ agent.start();
+ agent.put(event);
+ verify(source, times(1)).put(event);
+ }
+
+ @Test
+ public void testPutAll() throws EventDeliveryException {
+ Event event = new SimpleEvent();
+ List<Event> events = Lists.newArrayList();
+ events.add(event);
+ agent.configure(properties);
+ agent.start();
+ agent.putAll(events);
+ verify(source, times(1)).putAll(events);
+ }
+
+ @Test(expected = IllegalStateException.class)
+ public void testPutNotStarted() throws EventDeliveryException {
+ Event event = new SimpleEvent();
+ agent.configure(properties);
+ agent.put(event);
+ }
+
+ @Test(expected = IllegalStateException.class)
+ public void testPutAllNotStarted() throws EventDeliveryException {
+ Event event = new SimpleEvent();
+ List<Event> events = Lists.newArrayList();
+ events.add(event);
+ agent.configure(properties);
+ agent.putAll(events);
+ }
+}
diff --git
a/flume-embedded-agent/src/test/java/org/apache/flume/agent/embedded/TestEmbeddedAgentState.java
b/flume-embedded-agent/src/test/java/org/apache/flume/agent/embedded/TestEmbeddedAgentState.java
new file mode 100644
index 0000000..0f0ad23
--- /dev/null
+++
b/flume-embedded-agent/src/test/java/org/apache/flume/agent/embedded/TestEmbeddedAgentState.java
@@ -0,0 +1,125 @@
+/*
+ * 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.agent.embedded;
+
+import com.google.common.base.Throwables;
+import com.google.common.collect.Maps;
+import org.apache.flume.FlumeException;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.Map;
+
+public class TestEmbeddedAgentState {
+ private static final String HOSTNAME = "localhost";
+ private EmbeddedAgent agent;
+ private Map<String, String> properties;
+
+ @Before
+ public void setUp() throws Exception {
+ agent = new EmbeddedAgent("dummy");
+ properties = Maps.newHashMap();
+ properties.put("source.type",
EmbeddedAgentConfiguration.SOURCE_TYPE_EMBEDDED);
+ properties.put("channel.type", "memory");
+ properties.put("sinks", "sink1 sink2");
+ properties.put("sink1.type", "avro");
+ properties.put("sink2.type", "avro");
+ properties.put("sink1.hostname", HOSTNAME);
+ properties.put("sink1.port", "0");
+ properties.put("sink2.hostname", HOSTNAME);
+ properties.put("sink2.port", "0");
+ properties.put("processor.type", "load_balance");
+ }
+
+ @Test(expected = FlumeException.class)
+ public void testConfigureWithBadSourceType() {
+ properties.put(EmbeddedAgentConfiguration.SOURCE_TYPE, "bad");
+ agent.configure(properties);
+ }
+
+ @Test(expected = IllegalStateException.class)
+ public void testConfigureWhileStarted() {
+ try {
+ agent.configure(properties);
+ agent.start();
+ } catch (Exception e) {
+ Throwables.propagate(e);
+ }
+ agent.configure(properties);
+ }
+
+ @Test
+ public void testConfigureMultipleTimes() {
+ agent.configure(properties);
+ agent.configure(properties);
+ }
+
+ @Test(expected = IllegalStateException.class)
+ public void testStartWhileStarted() {
+ try {
+ agent.configure(properties);
+ agent.start();
+ } catch (Exception e) {
+ Throwables.propagate(e);
+ }
+ agent.start();
+ }
+
+ @Test(expected = IllegalStateException.class)
+ public void testStartUnconfigured() {
+ agent.start();
+ }
+
+ @Test(expected = IllegalStateException.class)
+ public void testStopBeforeConfigure() {
+ agent.stop();
+ }
+
+ @Test(expected = IllegalStateException.class)
+ public void testStoppedWhileStopped() {
+ try {
+ agent.configure(properties);
+ } catch (Exception e) {
+ Throwables.propagate(e);
+ }
+ agent.stop();
+ }
+
+ @Test(expected = IllegalStateException.class)
+ public void testStopAfterStop() {
+ try {
+ agent.configure(properties);
+ agent.start();
+ agent.stop();
+ } catch (Exception e) {
+ Throwables.propagate(e);
+ }
+ agent.stop();
+ }
+
+ @Test(expected = IllegalStateException.class)
+ public void testStopAfterConfigure() {
+ try {
+ agent.configure(properties);
+ } catch (Exception e) {
+ Throwables.propagate(e);
+ }
+ agent.stop();
+ }
+}
diff --git a/flume-embedded-agent/src/test/resources/log4j2.xml
b/flume-embedded-agent/src/test/resources/log4j2.xml
new file mode 100644
index 0000000..fbc44b4
--- /dev/null
+++ b/flume-embedded-agent/src/test/resources/log4j2.xml
@@ -0,0 +1,31 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You under the Apache License, Version 2.0
+ (the "License"); you may not use this file except in compliance with
+ the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+
+-->
+<Configuration status="OFF">
+ <Appenders>
+ <Console name="Console" target="SYSTEM_OUT">
+ <PatternLayout pattern="%d (%t) [%p - %l] %m%n" />
+ </Console>
+ </Appenders>
+ <Loggers>
+ <Logger name="org.apache.flume" level="DEBUG"/>
+ <Root level="INFO">
+ <AppenderRef ref="Console" />
+ </Root>
+ </Loggers>
+</Configuration>
\ No newline at end of file
diff --git a/pom.xml b/pom.xml
index e32ba4f..bee2b41 100644
--- a/pom.xml
+++ b/pom.xml
@@ -92,6 +92,7 @@ limitations under the License.
<modules>
<module>flume-rpc-avro</module>
<module>flume-rpc-thrift</module>
+ <module>flume-embedded-agent</module>
</modules>
<build>
<pluginManagement>
