You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@logging.apache.org by pk...@apache.org on 2026/10/05 15:33:53 UTC
(logging-flume) 01/01: Move the client API to a new `flume-client` artifact
This is an automated email from the ASF dual-hosted git repository.
ppkarwasz pushed a commit to branch feat/flume-client
in repository https://gitbox.apache.org/repos/asf/logging-flume.git
commit 7a1ad43f237bb568ab809bb5fa4dc115a4af6318
Author: Piotr P. Karwasz <pk...@apache.org>
AuthorDate: Mon Oct 5 17:33:35 2026 +0200
Move the client API to a new `flume-client` artifact
Client applications only need `Event`, the event builders and `RpcClient`, but had to depend on the SDK, which also carries the plugin API. Splitting them also removes a split package: `Event` moves to `org.apache.flume.event`, so the client artifact owns `org.apache.flume.api`, `org.apache.flume.event` and `org.apache.flume.util`.
`RpcClient` now extends `Closeable` and reports failures with `IOException` and the new `RpcDeliveryException`, so the client no longer needs the plugin-side `EventDeliveryException` and `FlumeException`. The event readers move to `org.apache.flume.serialization` and `EventHelper` to `org.apache.flume.core`, so that `org.apache.flume.event` lives in one artifact.
Assisted-By: Claude Opus 5.5 <no...@anthropic.com>
---
flume-bom/pom.xml | 5 ++
{flume-ng-sdk => flume-client}/pom.xml | 34 ++++------
.../org/apache/flume/api/AbstractRpcClient.java | 18 +++---
.../org/apache/flume/api/FailoverRpcClient.java | 75 ++++++++++++----------
.../main/java/org/apache/flume/api/HostInfo.java | 5 +-
.../apache/flume/api/LoadBalancingRpcClient.java | 30 ++++-----
.../main/java/org/apache/flume/api/RpcClient.java | 29 +++++----
.../flume/api/RpcClientConfigurationConstants.java | 0
.../org/apache/flume/api/RpcClientFactory.java | 42 ++++++------
.../org/apache/flume/api/RpcDeliveryException.java | 33 +++++-----
.../api/SSLContextAwareAbstractRpcClient.java | 3 +-
.../main/java/org/apache/flume/event}/Event.java | 2 +-
.../java/org/apache/flume/event/EventBuilder.java | 1 -
.../java/org/apache/flume/event/JSONEvent.java | 4 +-
.../java/org/apache/flume/event/SimpleEvent.java | 1 -
.../java/org/apache/flume/util/OrderSelector.java | 0
.../org/apache/flume/util/RandomOrderSelector.java | 0
.../apache/flume/util/RoundRobinOrderSelector.java | 0
.../main/java/org/apache/flume/util/SSLUtil.java | 0
.../apache/flume/util/SpecificOrderIterator.java | 0
.../org/apache/flume/event/TestEventBuilder.java | 4 +-
.../apache/flume/util/AbstractSSLUtilListTest.java | 0
.../org/apache/flume/util/AbstractSSLUtilTest.java | 0
.../flume/util/SSLUtilExcludeCipherSuitesTest.java | 0
.../flume/util/SSLUtilExcludeProtocolsTest.java | 0
.../flume/util/SSLUtilIncludeCipherSuitesTest.java | 0
.../flume/util/SSLUtilIncludeProtocolsTest.java | 0
.../flume/util/SSLUtilKeystorePasswordTest.java | 0
.../apache/flume/util/SSLUtilKeystorePathTest.java | 0
.../apache/flume/util/SSLUtilKeystoreTypeTest.java | 0
.../util/SSLUtilKeystoreTypeWithDefaultTest.java | 0
.../flume/util/SSLUtilTruststorePasswordTest.java | 0
.../flume/util/SSLUtilTruststorePathTest.java | 0
.../flume/util/SSLUtilTruststoreTypeTest.java | 0
.../util/SSLUtilTruststoreTypeWithDefaultTest.java | 0
flume-client/src/test/resources/log4j2.xml | 30 +++++++++
.../org/apache/flume/channel/file/EventUtils.java | 2 +-
.../org/apache/flume/channel/file/FileChannel.java | 2 +-
.../org/apache/flume/channel/file/FlumeEvent.java | 2 +-
.../java/org/apache/flume/channel/file/Log.java | 2 +-
.../instrumentation/FileChannelCounterMBean.java | 2 +-
.../apache/flume/channel/file/TestEventUtils.java | 2 +-
.../apache/flume/channel/file/TestFileChannel.java | 2 +-
.../org/apache/flume/channel/file/TestUtils.java | 2 +-
.../flume/channel/SpillableMemoryChannel.java | 2 +-
.../flume/channel/TestSpillableMemoryChannel.java | 2 +-
.../src/main/java/org/apache/flume/Channel.java | 1 +
.../java/org/apache/flume/ChannelSelector.java | 1 +
.../flume/channel/BasicChannelSemantics.java | 2 +-
.../flume/channel/BasicTransactionSemantics.java | 2 +-
.../org/apache/flume/channel/ChannelProcessor.java | 2 +-
.../org/apache/flume/channel/ChannelUtils.java | 2 +-
.../channel/LoadBalancingChannelSelector.java | 2 +-
.../org/apache/flume/channel/MemoryChannel.java | 2 +-
.../flume/channel/MultiplexingChannelSelector.java | 2 +-
.../flume/channel/PseudoTxnMemoryChannel.java | 2 +-
.../flume/channel/ReplicatingChannelSelector.java | 2 +-
.../channel/RoutableProxyChannelSelector.java | 2 +-
.../apache/flume/{event => core}/EventHelper.java | 4 +-
.../flume/formatter/output/EventFormatter.java | 2 +-
.../output/TextDelimitedOutputFormatter.java | 2 +-
.../apache/flume/interceptor/HostInterceptor.java | 2 +-
.../org/apache/flume/interceptor/Interceptor.java | 2 +-
.../apache/flume/interceptor/InterceptorChain.java | 2 +-
.../interceptor/RegexExtractorInterceptor.java | 2 +-
.../interceptor/RegexFilteringInterceptor.java | 2 +-
.../flume/interceptor/RemoveHeaderInterceptor.java | 4 +-
.../interceptor/SearchAndReplaceInterceptor.java | 2 +-
.../flume/interceptor/StaticInterceptor.java | 2 +-
.../flume/interceptor/TimestampInterceptor.java | 2 +-
.../serialization/BodyTextEventSerializer.java | 2 +-
.../flume/serialization/EventSerializer.java | 2 +-
.../HeaderAndBodyTextEventSerializer.java | 2 +-
.../org/apache/flume/sink/AbstractRpcSink.java | 8 ++-
.../java/org/apache/flume/sink/LoggerSink.java | 4 +-
.../main/java/org/apache/flume/sink/NullSink.java | 2 +-
.../org/apache/flume/sink/RollingFileSink.java | 2 +-
.../java/org/apache/flume/source/ExecSource.java | 2 +-
.../java/org/apache/flume/source/NetcatSource.java | 2 +-
.../flume/source/SequenceGeneratorSource.java | 2 +-
.../java/org/apache/flume/source/StressSource.java | 2 +-
.../channel/AbstractBasicChannelSemanticsTest.java | 2 +-
.../java/org/apache/flume/channel/MockChannel.java | 2 +-
.../java/org/apache/flume/channel/MockEvent.java | 2 +-
.../flume/channel/TestBasicChannelSemantics.java | 2 +-
.../apache/flume/channel/TestChannelProcessor.java | 2 +-
.../org/apache/flume/channel/TestChannelUtils.java | 2 +-
.../apache/flume/channel/TestMemoryChannel.java | 2 +-
.../channel/TestMemoryChannelConcurrency.java | 2 +-
.../channel/TestMemoryChannelTransaction.java | 2 +-
.../channel/TestMultiplexingChannelSelector.java | 2 +-
.../channel/TestRoutableProxyChannelSelector.java | 2 +-
.../flume/{event => core}/TestEventHelper.java | 3 +-
.../flume/interceptor/CensoringInterceptor.java | 2 +-
.../interceptor/RemoveHeaderInterceptorTest.java | 2 +-
.../interceptor/TestCensoringInterceptor.java | 2 +-
.../flume/interceptor/TestHostInterceptor.java | 2 +-
.../interceptor/TestRegexExtractorInterceptor.java | 2 +-
.../interceptor/TestRegexFilteringInterceptor.java | 2 +-
.../TestSearchAndReplaceInterceptor.java | 2 +-
.../flume/interceptor/TestStaticInterceptor.java | 2 +-
.../interceptor/TestTimestampInterceptor.java | 2 +-
.../flume/serialization/TestLineDeserializer.java | 2 +-
.../flume/sink/TestFailoverSinkProcessor.java | 2 +-
.../flume/sink/TestLoadBalancingSinkProcessor.java | 2 +-
.../java/org/apache/flume/sink/TestLoggerSink.java | 2 +-
.../org/apache/flume/sink/TestRollingFileSink.java | 2 +-
.../org/apache/flume/source/TestExecSource.java | 2 +-
.../org/apache/flume/source/TestNetcatSource.java | 2 +-
.../flume/source/TestPollableSourceRunner.java | 2 +-
.../flume/source/TestSequenceGeneratorSource.java | 2 +-
.../org/apache/flume/source/TestStressSource.java | 2 +-
.../node/TestAbstractConfigurationProvider.java | 2 +-
.../org/apache/flume/node/TestApplication.java | 2 +-
.../java/org/apache/flume/sink/NullInitSink.java | 2 +-
.../org/apache/flume/source/EventProcessor.java | 2 +-
.../java/org/apache/flume/source/LocalSource.java | 2 +-
.../org/apache/flume/source/TestNetcatSource.java | 2 +-
flume-ng-sdk/pom.xml | 9 ++-
.../flume/serialization/EventDeserializer.java | 2 +-
.../{event => serialization}/EventReader.java | 4 +-
.../flume/serialization/LineDeserializer.java | 2 +-
.../ReliableEventReader.java | 2 +-
.../SimpleTextLineEventReader.java | 5 +-
.../TestBufferedLineReader.java | 4 +-
.../flume/source/netcat/NetcatUdpSource.java | 2 +-
.../flume/source/netcat/TestNetcatUdpSource.java | 2 +-
.../source/syslog/MultiportSyslogTCPSource.java | 2 +-
.../apache/flume/source/syslog/SyslogParser.java | 2 +-
.../flume/source/syslog/SyslogTcpSource.java | 2 +-
.../flume/source/syslog/SyslogUDPSource.java | 2 +-
.../apache/flume/source/syslog/SyslogUtils.java | 2 +-
.../syslog/TestMultiportSyslogTCPSource.java | 2 +-
.../flume/source/syslog/TestSyslogParser.java | 2 +-
.../flume/source/syslog/TestSyslogTcpSource.java | 2 +-
.../flume/source/syslog/TestSyslogUdpSource.java | 2 +-
.../flume/source/syslog/TestSyslogUtils.java | 2 +-
.../source/taildir/ReliableTaildirEventReader.java | 4 +-
.../org/apache/flume/source/taildir/TailFile.java | 2 +-
.../apache/flume/source/taildir/TaildirSource.java | 2 +-
.../source/taildir/TestTaildirEventReader.java | 2 +-
.../flume/source/taildir/TestTaildirSource.java | 2 +-
.../org/apache/flume/tools/EventValidator.java | 2 +-
.../flume/tools/FileChannelIntegrityTool.java | 2 +-
.../flume/tools/TestFileChannelIntegrityTool.java | 2 +-
pom.xml | 1 +
src/changelog/.2.x.x/add_flume_client.xml | 7 ++
.../modules/ROOT/pages/FlumeDeveloperGuide.adoc | 40 ++++++------
148 files changed, 330 insertions(+), 279 deletions(-)
diff --git a/flume-bom/pom.xml b/flume-bom/pom.xml
index 608bc8a8..50c53a67 100644
--- a/flume-bom/pom.xml
+++ b/flume-bom/pom.xml
@@ -45,6 +45,11 @@
<artifactId>flume-auth</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.flume</groupId>
+ <artifactId>flume-client</artifactId>
+ <version>${project.version}</version>
+ </dependency>
<dependency>
<groupId>org.apache.flume</groupId>
<artifactId>flume-config-filter-api</artifactId>
diff --git a/flume-ng-sdk/pom.xml b/flume-client/pom.xml
similarity index 69%
copy from flume-ng-sdk/pom.xml
copy to flume-client/pom.xml
index b46565d8..122103d7 100644
--- a/flume-ng-sdk/pom.xml
+++ b/flume-client/pom.xml
@@ -16,6 +16,7 @@
~ 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>
@@ -25,36 +26,30 @@
<relativePath>../flume-parent</relativePath>
</parent>
- <artifactId>flume-sdk</artifactId>
- <name>Flume SDK</name>
- <description>Flume Software Development Kit: Stable public API for integration with Flume 1.x</description>
+ <artifactId>flume-client</artifactId>
+ <name>Flume Client</name>
+ <description>Client API to send events to a Flume agent.</description>
<properties>
- <!-- Baseline against the 1.x coordinates. TODO: remove once 2.0.0 is released. -->
- <flume.baseline.groupId>org.apache.flume</flume.baseline.groupId>
- <flume.baseline.artifactId>flume-ng-sdk</flume.baseline.artifactId>
+ <!-- New in 2.0.0: there is no previous release to check API compatibility against. -->
+ <bnd.baseline.fail.on.missing>false</bnd.baseline.fail.on.missing>
<!-- TODO fix spotbugs/pmd violations -->
- <spotbugs.maxAllowedViolations>69</spotbugs.maxAllowedViolations>
- <pmd.maxAllowedViolations>30</pmd.maxAllowedViolations>
- <module.name>org.apache.flume.sdk</module.name>
+ <spotbugs.maxAllowedViolations>26</spotbugs.maxAllowedViolations>
+ <pmd.maxAllowedViolations>10</pmd.maxAllowedViolations>
+ <module.name>org.apache.flume.client</module.name>
</properties>
<dependencies>
- <dependency>
- <groupId>junit</groupId>
- <artifactId>junit</artifactId>
- <scope>test</scope>
- </dependency>
-
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-api</artifactId>
</dependency>
<dependency>
- <groupId>com.google.guava</groupId>
- <artifactId>guava</artifactId>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <scope>test</scope>
</dependency>
<dependency>
@@ -63,11 +58,6 @@
<scope>test</scope>
</dependency>
- <dependency>
- <groupId>org.apache.commons</groupId>
- <artifactId>commons-lang3</artifactId>
- </dependency>
-
</dependencies>
</project>
diff --git a/flume-ng-sdk/src/main/java/org/apache/flume/api/AbstractRpcClient.java b/flume-client/src/main/java/org/apache/flume/api/AbstractRpcClient.java
similarity index 86%
rename from flume-ng-sdk/src/main/java/org/apache/flume/api/AbstractRpcClient.java
rename to flume-client/src/main/java/org/apache/flume/api/AbstractRpcClient.java
index 12043e2b..6e2f04f5 100644
--- a/flume-ng-sdk/src/main/java/org/apache/flume/api/AbstractRpcClient.java
+++ b/flume-client/src/main/java/org/apache/flume/api/AbstractRpcClient.java
@@ -16,11 +16,10 @@
*/
package org.apache.flume.api;
+import java.io.IOException;
import java.util.List;
import java.util.Properties;
-import org.apache.flume.Event;
-import org.apache.flume.EventDeliveryException;
-import org.apache.flume.FlumeException;
+import org.apache.flume.event.Event;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@@ -37,24 +36,25 @@ public abstract class AbstractRpcClient implements RpcClient {
}
@Override
- public abstract void append(Event event) throws EventDeliveryException;
+ public abstract void append(Event event) throws RpcDeliveryException;
@Override
- public abstract void appendBatch(List<Event> events) throws EventDeliveryException;
+ public abstract void appendBatch(List<Event> events) throws RpcDeliveryException;
@Override
public abstract boolean isActive();
@Override
- public abstract void close() throws FlumeException;
+ public abstract void close() throws IOException;
/**
* Configure the client using the given properties object.
* @param properties
- * @throws FlumeException if the client can not be configured using this
- * method, or if the client was already configured once.
+ * @throws IllegalArgumentException if the properties are invalid.
+ * @throws IllegalStateException if the client was already configured once.
+ * @throws IOException if the client fails to connect.
*/
- protected abstract void configure(Properties properties) throws FlumeException;
+ protected abstract void configure(Properties properties) throws IOException;
/**
* This is to parse the batch size config for rpc clients
diff --git a/flume-ng-sdk/src/main/java/org/apache/flume/api/FailoverRpcClient.java b/flume-client/src/main/java/org/apache/flume/api/FailoverRpcClient.java
similarity index 81%
rename from flume-ng-sdk/src/main/java/org/apache/flume/api/FailoverRpcClient.java
rename to flume-client/src/main/java/org/apache/flume/api/FailoverRpcClient.java
index 8d03bb73..7a6fb121 100644
--- a/flume-ng-sdk/src/main/java/org/apache/flume/api/FailoverRpcClient.java
+++ b/flume-client/src/main/java/org/apache/flume/api/FailoverRpcClient.java
@@ -16,12 +16,11 @@
*/
package org.apache.flume.api;
+import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.List;
import java.util.Properties;
-import org.apache.flume.Event;
-import org.apache.flume.EventDeliveryException;
-import org.apache.flume.FlumeException;
+import org.apache.flume.event.Event;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@@ -62,10 +61,10 @@ public class FailoverRpcClient extends AbstractRpcClient implements RpcClient {
// This function has to be synchronized to establish a happens-before
// relationship for different threads that access this object
// since shared data structures are created here.
- private synchronized void configureHosts(Properties properties) throws FlumeException {
+ private synchronized void configureHosts(Properties properties) {
if (isActive) {
logger.error("This client was already configured, " + "cannot reconfigure.");
- throw new FlumeException("This client was already configured, " + "cannot reconfigure.");
+ throw new IllegalStateException("This client was already configured, " + "cannot reconfigure.");
}
hosts = HostInfo.getHostInfoList(properties);
String tries = properties.getProperty(RpcClientConfigurationConstants.CONFIG_MAX_ATTEMPTS);
@@ -95,7 +94,7 @@ public class FailoverRpcClient extends AbstractRpcClient implements RpcClient {
return maxTries;
}
- private synchronized RpcClient getClient() {
+ private synchronized RpcClient getClient() throws RpcDeliveryException {
if (client == null || !this.client.isActive()) {
client = getNextClient();
return client;
@@ -110,41 +109,42 @@ public class FailoverRpcClient extends AbstractRpcClient implements RpcClient {
*
* @param event The event to be appended.
*
- * @throws EventDeliveryException
+ * @throws RpcDeliveryException
*/
@Override
- public void append(Event event) throws EventDeliveryException {
+ public void append(Event event) throws RpcDeliveryException {
// Why a local variable rather than just calling getClient()?
- // If we get an EventDeliveryException, we need to call close on
+ // If we get an RpcDeliveryException, we need to call close on
// that specific client, getClient in this case, will get us
// the next client - leaving a resource leak.
RpcClient localClient = null;
synchronized (this) {
if (!isActive) {
logger.error("Attempting to append to an already closed client.");
- throw new EventDeliveryException("Attempting to append to an already closed client.");
+ throw new RpcDeliveryException("Attempting to append to an already closed client.");
}
}
// Sit in an infinite loop and try to append!
int tries = 0;
while (tries < maxTries) {
+ tries++;
+ // Fails without retrying if no host is reachable.
+ localClient = getClient();
try {
- tries++;
- localClient = getClient();
localClient.append(event);
return;
- } catch (EventDeliveryException e) {
+ } catch (RpcDeliveryException e) {
// Could not send event through this client, try to pick another client.
logger.warn("Client failed. Exception follows: ", e);
- localClient.close();
+ closeQuietly(localClient);
localClient = null;
- } catch (Exception e2) {
+ } catch (RuntimeException e2) {
logger.error("Failed to send event: ", e2);
- throw new EventDeliveryException("Failed to send event. Exception follows: ", e2);
+ throw new RpcDeliveryException("Failed to send event. Exception follows: ", e2);
}
}
logger.error("Tried many times, could not send event.");
- throw new EventDeliveryException("Failed to send the event!");
+ throw new RpcDeliveryException("Failed to send the event!");
}
/**
@@ -153,36 +153,37 @@ public class FailoverRpcClient extends AbstractRpcClient implements RpcClient {
*
* @param events The events to be appended.
*
- * @throws EventDeliveryException
+ * @throws RpcDeliveryException
*/
@Override
- public void appendBatch(List<Event> events) throws EventDeliveryException {
+ public void appendBatch(List<Event> events) throws RpcDeliveryException {
RpcClient localClient = null;
synchronized (this) {
if (!isActive) {
logger.error("Attempting to append to an already closed client.");
- throw new EventDeliveryException("Attempting to append to an already closed client!");
+ throw new RpcDeliveryException("Attempting to append to an already closed client!");
}
}
int tries = 0;
while (tries < maxTries) {
+ tries++;
+ // Fails without retrying if no host is reachable.
+ localClient = getClient();
try {
- tries++;
- localClient = getClient();
localClient.appendBatch(events);
return;
- } catch (EventDeliveryException e) {
+ } catch (RpcDeliveryException e) {
// Could not send event through this client, try to pick another client.
logger.warn("Client failed. Exception follows: ", e);
- localClient.close();
+ closeQuietly(localClient);
localClient = null;
- } catch (Exception e1) {
+ } catch (RuntimeException e1) {
logger.error("No clients active: ", e1);
- throw new EventDeliveryException("No clients currently active. " + "Exception follows: ", e1);
+ throw new RpcDeliveryException("No clients currently active. " + "Exception follows: ", e1);
}
}
logger.error("Tried many times, could not send event.");
- throw new EventDeliveryException("Failed to send the event!");
+ throw new RpcDeliveryException("Failed to send the event!");
}
// Returns false if and only if this client has been closed explicitly.
@@ -197,7 +198,7 @@ public class FailoverRpcClient extends AbstractRpcClient implements RpcClient {
* Close the connection. This function is safe to call over and over.
*/
@Override
- public synchronized void close() throws FlumeException {
+ public synchronized void close() throws IOException {
if (client != null) {
client.close();
isActive = false;
@@ -215,7 +216,15 @@ public class FailoverRpcClient extends AbstractRpcClient implements RpcClient {
return new InetSocketAddress(hostInfo.getHostName(), hostInfo.getPortNumber());
}
- private RpcClient getNextClient() throws FlumeException {
+ private static void closeQuietly(RpcClient client) {
+ try {
+ client.close();
+ } catch (IOException e) {
+ logger.warn("Failed to close client.", e);
+ }
+ }
+
+ private RpcClient getNextClient() throws RpcDeliveryException {
lastCheckedhost = (lastCheckedhost == (hosts.size() - 1)) ? -1 : lastCheckedhost;
RpcClient localClient = null;
int limit = hosts.size();
@@ -233,7 +242,7 @@ public class FailoverRpcClient extends AbstractRpcClient implements RpcClient {
localClient = RpcClientFactory.getInstance(props);
lastCheckedhost = count;
return localClient;
- } catch (FlumeException e) {
+ } catch (IOException e) {
logger.info("Could not connect to " + hostInfo, e);
continue;
}
@@ -245,7 +254,7 @@ public class FailoverRpcClient extends AbstractRpcClient implements RpcClient {
localClient = RpcClientFactory.getInstance(props);
lastCheckedhost = count;
return localClient;
- } catch (FlumeException e) {
+ } catch (IOException e) {
logger.info("Could not connect to " + hostInfo, e);
continue;
}
@@ -253,7 +262,7 @@ public class FailoverRpcClient extends AbstractRpcClient implements RpcClient {
if (localClient == null) {
lastCheckedhost = -1;
logger.error("No active client found.");
- throw new FlumeException("No active client.");
+ throw new RpcDeliveryException("No active client.");
}
// This return should never be reached!
return localClient;
@@ -265,7 +274,7 @@ public class FailoverRpcClient extends AbstractRpcClient implements RpcClient {
}
@Override
- public void configure(Properties properties) throws FlumeException {
+ public void configure(Properties properties) {
configurationProperties = new Properties();
configurationProperties.putAll(properties);
diff --git a/flume-ng-sdk/src/main/java/org/apache/flume/api/HostInfo.java b/flume-client/src/main/java/org/apache/flume/api/HostInfo.java
similarity index 93%
rename from flume-ng-sdk/src/main/java/org/apache/flume/api/HostInfo.java
rename to flume-client/src/main/java/org/apache/flume/api/HostInfo.java
index 9e4aa52c..b1006ee3 100644
--- a/flume-ng-sdk/src/main/java/org/apache/flume/api/HostInfo.java
+++ b/flume-client/src/main/java/org/apache/flume/api/HostInfo.java
@@ -19,7 +19,6 @@ package org.apache.flume.api;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
-import org.apache.flume.FlumeException;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@@ -74,14 +73,14 @@ public class HostInfo {
String[] hostAndPort = hostAndPortStr.split(":");
if (hostAndPort.length != 2) {
logger.error("Invalid host address" + hostAndPortStr);
- throw new FlumeException("Invalid host address" + hostAndPortStr);
+ throw new IllegalArgumentException("Invalid host address" + hostAndPortStr);
}
Integer port = null;
try {
port = Integer.parseInt(hostAndPort[1]);
} catch (NumberFormatException e) {
logger.error("Invalid port number" + hostAndPortStr, e);
- throw new FlumeException("Invalid port number" + hostAndPortStr);
+ throw new IllegalArgumentException("Invalid port number" + hostAndPortStr);
}
HostInfo info = new HostInfo(hostList[i], hostAndPort[0].trim(), port);
hosts.add(info);
diff --git a/flume-ng-sdk/src/main/java/org/apache/flume/api/LoadBalancingRpcClient.java b/flume-client/src/main/java/org/apache/flume/api/LoadBalancingRpcClient.java
similarity index 90%
rename from flume-ng-sdk/src/main/java/org/apache/flume/api/LoadBalancingRpcClient.java
rename to flume-client/src/main/java/org/apache/flume/api/LoadBalancingRpcClient.java
index c12a0a4f..0b246d31 100644
--- a/flume-ng-sdk/src/main/java/org/apache/flume/api/LoadBalancingRpcClient.java
+++ b/flume-client/src/main/java/org/apache/flume/api/LoadBalancingRpcClient.java
@@ -16,14 +16,13 @@
*/
package org.apache.flume.api;
+import java.io.IOException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Properties;
-import org.apache.flume.Event;
-import org.apache.flume.EventDeliveryException;
-import org.apache.flume.FlumeException;
+import org.apache.flume.event.Event;
import org.apache.flume.util.OrderSelector;
import org.apache.flume.util.RandomOrderSelector;
import org.apache.flume.util.RoundRobinOrderSelector;
@@ -56,7 +55,7 @@ public class LoadBalancingRpcClient extends AbstractRpcClient {
private volatile boolean isOpen = false;
@Override
- public void append(Event event) throws EventDeliveryException {
+ public void append(Event event) throws RpcDeliveryException {
throwIfClosed();
boolean eventSent = false;
Iterator<HostInfo> it = selector.createHostIterator();
@@ -75,12 +74,12 @@ public class LoadBalancingRpcClient extends AbstractRpcClient {
}
if (!eventSent) {
- throw new EventDeliveryException("Unable to send event to any host");
+ throw new RpcDeliveryException("Unable to send event to any host");
}
}
@Override
- public void appendBatch(List<Event> events) throws EventDeliveryException {
+ public void appendBatch(List<Event> events) throws RpcDeliveryException {
throwIfClosed();
boolean batchSent = false;
Iterator<HostInfo> it = selector.createHostIterator();
@@ -99,7 +98,7 @@ public class LoadBalancingRpcClient extends AbstractRpcClient {
}
if (!batchSent) {
- throw new EventDeliveryException("Unable to send batch to any host");
+ throw new RpcDeliveryException("Unable to send batch to any host");
}
}
@@ -108,14 +107,14 @@ public class LoadBalancingRpcClient extends AbstractRpcClient {
return isOpen;
}
- private void throwIfClosed() throws EventDeliveryException {
+ private void throwIfClosed() throws RpcDeliveryException {
if (!isOpen) {
- throw new EventDeliveryException("Rpc Client is closed");
+ throw new RpcDeliveryException("Rpc Client is closed");
}
}
@Override
- public void close() throws FlumeException {
+ public void close() {
isOpen = false;
synchronized (this) {
Iterator<String> it = clientMap.keySet().iterator();
@@ -135,13 +134,14 @@ public class LoadBalancingRpcClient extends AbstractRpcClient {
}
@Override
- protected void configure(Properties properties) throws FlumeException {
+ protected void configure(Properties properties) {
clientMap = new HashMap<String, RpcClient>();
configurationProperties = new Properties();
configurationProperties.putAll(properties);
hosts = HostInfo.getHostInfoList(properties);
if (hosts.size() < 2) {
- throw new FlumeException("At least two hosts are required to use the " + "load balancing RPC client.");
+ throw new IllegalArgumentException(
+ "At least two hosts are required to use the " + "load balancing RPC client.");
}
String lbTypeName = properties.getProperty(
@@ -169,7 +169,7 @@ public class LoadBalancingRpcClient extends AbstractRpcClient {
selector = klass.newInstance();
} catch (Exception ex) {
- throw new FlumeException("Unable to instantiate host selector: " + lbTypeName, ex);
+ throw new IllegalArgumentException("Unable to instantiate host selector: " + lbTypeName, ex);
}
}
@@ -178,7 +178,7 @@ public class LoadBalancingRpcClient extends AbstractRpcClient {
isOpen = true;
}
- private synchronized RpcClient getClient(HostInfo info) throws FlumeException, EventDeliveryException {
+ private synchronized RpcClient getClient(HostInfo info) throws IOException {
throwIfClosed();
String name = info.getReferenceName();
RpcClient client = clientMap.get(name);
@@ -198,7 +198,7 @@ public class LoadBalancingRpcClient extends AbstractRpcClient {
return client;
}
- private RpcClient createClient(String referenceName) throws FlumeException {
+ private RpcClient createClient(String referenceName) throws IOException {
Properties props = getClientConfigurationProperties(referenceName);
return RpcClientFactory.getInstance(props);
}
diff --git a/flume-ng-sdk/src/main/java/org/apache/flume/api/RpcClient.java b/flume-client/src/main/java/org/apache/flume/api/RpcClient.java
similarity index 80%
rename from flume-ng-sdk/src/main/java/org/apache/flume/api/RpcClient.java
rename to flume-client/src/main/java/org/apache/flume/api/RpcClient.java
index 87d63dfe..89361e4a 100644
--- a/flume-ng-sdk/src/main/java/org/apache/flume/api/RpcClient.java
+++ b/flume-client/src/main/java/org/apache/flume/api/RpcClient.java
@@ -16,16 +16,14 @@
*/
package org.apache.flume.api;
+import java.io.Closeable;
+import java.io.IOException;
import java.util.List;
-import org.apache.flume.Event;
-import org.apache.flume.EventDeliveryException;
-import org.apache.flume.FlumeException;
+import org.apache.flume.event.Event;
/**
* <p>Public client interface for sending data to Flume.</p>
*
- * <p>This interface is intended not to change incompatibly for Flume 1.x.</p>
- *
* <p><strong>Note:</strong> It is recommended for applications to construct
* {@link RpcClient} instances using the {@link RpcClientFactory} class,
* instead of using builders associated with a particular implementation class.
@@ -33,7 +31,7 @@ import org.apache.flume.FlumeException;
*
* @see org.apache.flume.api.RpcClientFactory
*/
-public interface RpcClient {
+public interface RpcClient extends Closeable {
/**
* Returns the maximum number of {@link Event events} that may be batched
@@ -48,15 +46,15 @@ public interface RpcClient {
* </p>
*
* <p><strong>Note:</strong> If this method throws an
- * {@link EventDeliveryException}, there is no way to recover and the
+ * {@link RpcDeliveryException}, there is no way to recover and the
* application must invoke {@link #close()} on this object to clean up system
* resources.</p>
*
* @param event
*
- * @throws EventDeliveryException when an error prevents event delivery.
+ * @throws RpcDeliveryException when an error prevents event delivery.
*/
- public void append(Event event) throws EventDeliveryException;
+ public void append(Event event) throws RpcDeliveryException;
/**
* <p>Send a list of {@linkplain Event events} to the associated Flume source.
@@ -71,15 +69,15 @@ public interface RpcClient {
* increase.</p>
*
* <p><strong>Note:</strong> If this method throws an
- * {@link EventDeliveryException}, there is no way to recover and the
+ * {@link RpcDeliveryException}, there is no way to recover and the
* application must invoke {@link #close()} on this object to clean up system
* resources.</p>
*
* @param events List of events to send
*
- * @throws EventDeliveryException when an error prevents event delivery.
+ * @throws RpcDeliveryException when an error prevents event delivery.
*/
- public void appendBatch(List<Event> events) throws EventDeliveryException;
+ public void appendBatch(List<Event> events) throws RpcDeliveryException;
/**
* <p>Returns {@code true} if this object appears to be in a usable state, and
@@ -98,8 +96,11 @@ public interface RpcClient {
*
* <p>Multi-threaded applications may want to gracefully stop making
* RPC calls before calling close(). Otherwise, they risk getting
- * {@link EventDeliveryException} thrown from their in-flight calls when the
+ * {@link RpcDeliveryException} thrown from their in-flight calls when the
* underlying connection is disabled.</p>
+ *
+ * @throws IOException if the underlying connection fails to close.
*/
- public void close() throws FlumeException;
+ @Override
+ public void close() throws IOException;
}
diff --git a/flume-ng-sdk/src/main/java/org/apache/flume/api/RpcClientConfigurationConstants.java b/flume-client/src/main/java/org/apache/flume/api/RpcClientConfigurationConstants.java
similarity index 100%
rename from flume-ng-sdk/src/main/java/org/apache/flume/api/RpcClientConfigurationConstants.java
rename to flume-client/src/main/java/org/apache/flume/api/RpcClientConfigurationConstants.java
diff --git a/flume-ng-sdk/src/main/java/org/apache/flume/api/RpcClientFactory.java b/flume-client/src/main/java/org/apache/flume/api/RpcClientFactory.java
similarity index 90%
rename from flume-ng-sdk/src/main/java/org/apache/flume/api/RpcClientFactory.java
rename to flume-client/src/main/java/org/apache/flume/api/RpcClientFactory.java
index 5a500cb6..1c65ce91 100644
--- a/flume-ng-sdk/src/main/java/org/apache/flume/api/RpcClientFactory.java
+++ b/flume-client/src/main/java/org/apache/flume/api/RpcClientFactory.java
@@ -17,14 +17,12 @@
package org.apache.flume.api;
import java.io.File;
-import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;
import java.io.Reader;
import java.lang.reflect.InvocationTargetException;
import java.util.Locale;
import java.util.Properties;
-import org.apache.flume.FlumeException;
/**
* Factory class to construct Flume {@link RPCClient} implementations.
@@ -50,10 +48,10 @@ public class RpcClientFactory {
* @see org.apache.flume.api.NettyAvroClient
*
* @param properties The properties to instantiate the client with.
- * @throws FlumeException
+ * @throws IOException if the client fails to connect.
*/
@SuppressWarnings("unchecked")
- public static RpcClient getInstance(Properties properties) throws FlumeException {
+ public static RpcClient getInstance(Properties properties) throws IOException {
String type = null;
type = properties.getProperty(RpcClientConfigurationConstants.CONFIG_CLIENT_TYPE);
if (type == null || type.isEmpty()) {
@@ -78,16 +76,17 @@ public class RpcClientFactory {
}
clazz = (Class<? extends AbstractRpcClient>) Class.forName(clientClassType);
} catch (ClassNotFoundException e) {
- throw new FlumeException("No such client!", e);
+ throw new IllegalArgumentException("No such client: " + type, e);
}
try {
client = clazz.newInstance();
} catch (InstantiationException e) {
- throw new FlumeException(
+ throw new IllegalArgumentException(
"Cannot instantiate client " + clazz.getCanonicalName() + " Exception follows:", e);
} catch (IllegalAccessException e) {
- throw new FlumeException("Cannot instantiate client " + clazz.getCanonicalName() + "Exception follows:", e);
+ throw new IllegalArgumentException(
+ "Cannot instantiate client " + clazz.getCanonicalName() + "Exception follows:", e);
}
client.configure(properties);
return client;
@@ -101,7 +100,7 @@ public class RpcClientFactory {
* @throws FileNotFoundException If the file cannot be found
* @throws IOException If there is an IO error
*/
- public static RpcClient getInstance(File propertiesFile) throws FileNotFoundException, IOException {
+ public static RpcClient getInstance(File propertiesFile) throws IOException {
Reader reader = new FileReader(propertiesFile);
Properties props = new Properties();
props.load(reader);
@@ -111,20 +110,20 @@ public class RpcClientFactory {
/**
* Deprecated. Use
* {@link #getDefaultInstance(String, Integer)} instead.
- * @throws FlumeException
+ * @throws IOException if the client fails to connect.
* @deprecated
*/
@Deprecated
- public static RpcClient getInstance(String hostname, Integer port) throws FlumeException {
+ public static RpcClient getInstance(String hostname, Integer port) throws IOException {
return getDefaultInstance(hostname, port);
}
/**
* Returns an instance of {@link RpcClient} connected to the specified
* {@code hostname} and {@code port}.
- * @throws FlumeException
+ * @throws IOException if the client fails to connect.
*/
- public static RpcClient getDefaultInstance(String hostname, Integer port) throws FlumeException {
+ public static RpcClient getDefaultInstance(String hostname, Integer port) throws IOException {
return getDefaultInstance(hostname, port, 0);
}
@@ -132,20 +131,20 @@ public class RpcClientFactory {
* Deprecated. Use
* {@link #getDefaultInstance(String, Integer, Integer)}
* instead.
- * @throws FlumeException
+ * @throws IOException if the client fails to connect.
* @deprecated
*/
@Deprecated
- public static RpcClient getInstance(String hostname, Integer port, Integer batchSize) throws FlumeException {
+ public static RpcClient getInstance(String hostname, Integer port, Integer batchSize) throws IOException {
return getDefaultInstance(hostname, port, batchSize);
}
/**
* Returns an instance of {@link RpcClient} connected to the specified
* {@code hostname} and {@code port} with the specified {@code batchSize}.
- * @throws FlumeException
+ * @throws IOException if the client fails to connect.
*/
- public static RpcClient getDefaultInstance(String hostname, Integer port, Integer batchSize) throws FlumeException {
+ public static RpcClient getDefaultInstance(String hostname, Integer port, Integer batchSize) throws IOException {
if (hostname == null) {
throw new NullPointerException("hostname must not be null");
@@ -174,7 +173,7 @@ public class RpcClientFactory {
* @return an {@linkplain RpcClient} which uses thrift configured with the
* given parameters.
*/
- public static RpcClient getThriftInstance(String hostname, Integer port, Integer batchSize) {
+ public static RpcClient getThriftInstance(String hostname, Integer port, Integer batchSize) throws IOException {
if (hostname == null) {
throw new NullPointerException("hostname must not be null");
}
@@ -192,7 +191,7 @@ public class RpcClientFactory {
return createClient(THRIFT_CLASS_NAME, props);
}
- private static RpcClient createClient(String className, Properties props) throws FlumeException {
+ private static RpcClient createClient(String className, Properties props) throws IOException {
try {
@SuppressWarnings("unchecked")
Class<? extends AbstractRpcClient> clazz = (Class<? extends AbstractRpcClient>) Class.forName(className);
@@ -204,7 +203,8 @@ public class RpcClientFactory {
| NoSuchMethodException
| IllegalAccessException
| InvocationTargetException e) {
- throw new FlumeException("Cannot instantiate client. Implementation " + className + " failed to load.", e);
+ throw new IllegalStateException(
+ "Cannot instantiate client. Implementation " + className + " failed to load.", e);
}
}
@@ -218,7 +218,7 @@ public class RpcClientFactory {
* @return - An {@linkplain RpcClient} which uses thrift configured with the
* given parameters.
*/
- public static RpcClient getThriftInstance(String hostname, Integer port) {
+ public static RpcClient getThriftInstance(String hostname, Integer port) throws IOException {
return getThriftInstance(hostname, port, RpcClientConfigurationConstants.DEFAULT_BATCH_SIZE);
}
@@ -229,7 +229,7 @@ public class RpcClientFactory {
* @return - An {@linkplain RpcClient} which uses thrift configured with the
* given parameters.
*/
- public static RpcClient getThriftInstance(Properties props) {
+ public static RpcClient getThriftInstance(Properties props) throws IOException {
props.setProperty(RpcClientConfigurationConstants.CONFIG_CLIENT_TYPE, ClientType.THRIFT.clientClassName);
return getInstance(props);
}
diff --git a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/EventUtils.java b/flume-client/src/main/java/org/apache/flume/api/RpcDeliveryException.java
similarity index 59%
copy from flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/EventUtils.java
copy to flume-client/src/main/java/org/apache/flume/api/RpcDeliveryException.java
index 11f52389..d8715f69 100644
--- a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/EventUtils.java
+++ b/flume-client/src/main/java/org/apache/flume/api/RpcDeliveryException.java
@@ -14,25 +14,28 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.flume.channel.file;
+package org.apache.flume.api;
-import org.apache.flume.Event;
+import java.io.IOException;
+import java.io.Serial;
/**
- *
+ * Signals that an {@link RpcClient} failed to deliver events to a Flume agent.
*/
-public class EventUtils {
+public class RpcDeliveryException extends IOException {
+
+ @Serial
+ private static final long serialVersionUID = -4237375912986132195L;
+
+ public RpcDeliveryException(String message) {
+ super(message);
+ }
+
+ public RpcDeliveryException(String message, Throwable cause) {
+ super(message, cause);
+ }
- /**
- * Returns the Event encapsulated by a Put wrapper
- *
- * @param transactionEventRecord TransactionEvent
- * @return Event if Put instance is present, null otherwise
- */
- public static Event getEventFromTransactionEvent(TransactionEventRecord transactionEventRecord) {
- if (transactionEventRecord instanceof Put) {
- return ((Put) transactionEventRecord).getEvent();
- }
- return null;
+ public RpcDeliveryException(Throwable cause) {
+ super(cause);
}
}
diff --git a/flume-ng-sdk/src/main/java/org/apache/flume/api/SSLContextAwareAbstractRpcClient.java b/flume-client/src/main/java/org/apache/flume/api/SSLContextAwareAbstractRpcClient.java
similarity index 96%
rename from flume-ng-sdk/src/main/java/org/apache/flume/api/SSLContextAwareAbstractRpcClient.java
rename to flume-client/src/main/java/org/apache/flume/api/SSLContextAwareAbstractRpcClient.java
index 78cd4f3c..7ffdb77b 100644
--- a/flume-ng-sdk/src/main/java/org/apache/flume/api/SSLContextAwareAbstractRpcClient.java
+++ b/flume-client/src/main/java/org/apache/flume/api/SSLContextAwareAbstractRpcClient.java
@@ -21,7 +21,6 @@ import java.util.LinkedHashSet;
import java.util.Objects;
import java.util.Properties;
import java.util.Set;
-import org.apache.flume.FlumeException;
import org.apache.flume.util.SSLUtil;
public abstract class SSLContextAwareAbstractRpcClient extends AbstractRpcClient {
@@ -35,7 +34,7 @@ public abstract class SSLContextAwareAbstractRpcClient extends AbstractRpcClient
protected final Set<String> excludeCipherSuites = new LinkedHashSet<>();
protected final Set<String> includeCipherSuites = new LinkedHashSet<>();
- protected void configureSSL(Properties properties) throws FlumeException {
+ protected void configureSSL(Properties properties) {
enableSsl = Boolean.parseBoolean(properties.getProperty(RpcClientConfigurationConstants.CONFIG_SSL));
trustAllCerts =
Boolean.parseBoolean(properties.getProperty(RpcClientConfigurationConstants.CONFIG_TRUST_ALL_CERTS));
diff --git a/flume-ng-sdk/src/main/java/org/apache/flume/Event.java b/flume-client/src/main/java/org/apache/flume/event/Event.java
similarity index 97%
rename from flume-ng-sdk/src/main/java/org/apache/flume/Event.java
rename to flume-client/src/main/java/org/apache/flume/event/Event.java
index c213cdaf..2963d52d 100644
--- a/flume-ng-sdk/src/main/java/org/apache/flume/Event.java
+++ b/flume-client/src/main/java/org/apache/flume/event/Event.java
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.flume;
+package org.apache.flume.event;
import java.util.Map;
diff --git a/flume-ng-sdk/src/main/java/org/apache/flume/event/EventBuilder.java b/flume-client/src/main/java/org/apache/flume/event/EventBuilder.java
similarity index 98%
rename from flume-ng-sdk/src/main/java/org/apache/flume/event/EventBuilder.java
rename to flume-client/src/main/java/org/apache/flume/event/EventBuilder.java
index 48551eb8..610a0b62 100644
--- a/flume-ng-sdk/src/main/java/org/apache/flume/event/EventBuilder.java
+++ b/flume-client/src/main/java/org/apache/flume/event/EventBuilder.java
@@ -19,7 +19,6 @@ package org.apache.flume.event;
import java.nio.charset.Charset;
import java.util.HashMap;
import java.util.Map;
-import org.apache.flume.Event;
public class EventBuilder {
diff --git a/flume-ng-sdk/src/main/java/org/apache/flume/event/JSONEvent.java b/flume-client/src/main/java/org/apache/flume/event/JSONEvent.java
similarity index 91%
rename from flume-ng-sdk/src/main/java/org/apache/flume/event/JSONEvent.java
rename to flume-client/src/main/java/org/apache/flume/event/JSONEvent.java
index e03a769e..8ddef7dd 100644
--- a/flume-ng-sdk/src/main/java/org/apache/flume/event/JSONEvent.java
+++ b/flume-client/src/main/java/org/apache/flume/event/JSONEvent.java
@@ -18,8 +18,6 @@ package org.apache.flume.event;
import java.io.UnsupportedEncodingException;
import java.util.Map;
-import org.apache.flume.Event;
-import org.apache.flume.FlumeException;
/**
*
@@ -45,7 +43,7 @@ public class JSONEvent implements Event {
try {
return body.getBytes(charset);
} catch (UnsupportedEncodingException ex) {
- throw new FlumeException(String.format("%s encoding not supported", charset), ex);
+ throw new IllegalArgumentException(String.format("%s encoding not supported", charset), ex);
}
} else {
return new byte[0];
diff --git a/flume-ng-sdk/src/main/java/org/apache/flume/event/SimpleEvent.java b/flume-client/src/main/java/org/apache/flume/event/SimpleEvent.java
similarity index 98%
rename from flume-ng-sdk/src/main/java/org/apache/flume/event/SimpleEvent.java
rename to flume-client/src/main/java/org/apache/flume/event/SimpleEvent.java
index b56008ea..64e7e14c 100644
--- a/flume-ng-sdk/src/main/java/org/apache/flume/event/SimpleEvent.java
+++ b/flume-client/src/main/java/org/apache/flume/event/SimpleEvent.java
@@ -18,7 +18,6 @@ package org.apache.flume.event;
import java.util.HashMap;
import java.util.Map;
-import org.apache.flume.Event;
public class SimpleEvent implements Event {
diff --git a/flume-ng-sdk/src/main/java/org/apache/flume/util/OrderSelector.java b/flume-client/src/main/java/org/apache/flume/util/OrderSelector.java
similarity index 100%
rename from flume-ng-sdk/src/main/java/org/apache/flume/util/OrderSelector.java
rename to flume-client/src/main/java/org/apache/flume/util/OrderSelector.java
diff --git a/flume-ng-sdk/src/main/java/org/apache/flume/util/RandomOrderSelector.java b/flume-client/src/main/java/org/apache/flume/util/RandomOrderSelector.java
similarity index 100%
rename from flume-ng-sdk/src/main/java/org/apache/flume/util/RandomOrderSelector.java
rename to flume-client/src/main/java/org/apache/flume/util/RandomOrderSelector.java
diff --git a/flume-ng-sdk/src/main/java/org/apache/flume/util/RoundRobinOrderSelector.java b/flume-client/src/main/java/org/apache/flume/util/RoundRobinOrderSelector.java
similarity index 100%
rename from flume-ng-sdk/src/main/java/org/apache/flume/util/RoundRobinOrderSelector.java
rename to flume-client/src/main/java/org/apache/flume/util/RoundRobinOrderSelector.java
diff --git a/flume-ng-sdk/src/main/java/org/apache/flume/util/SSLUtil.java b/flume-client/src/main/java/org/apache/flume/util/SSLUtil.java
similarity index 100%
rename from flume-ng-sdk/src/main/java/org/apache/flume/util/SSLUtil.java
rename to flume-client/src/main/java/org/apache/flume/util/SSLUtil.java
diff --git a/flume-ng-sdk/src/main/java/org/apache/flume/util/SpecificOrderIterator.java b/flume-client/src/main/java/org/apache/flume/util/SpecificOrderIterator.java
similarity index 100%
rename from flume-ng-sdk/src/main/java/org/apache/flume/util/SpecificOrderIterator.java
rename to flume-client/src/main/java/org/apache/flume/util/SpecificOrderIterator.java
diff --git a/flume-ng-sdk/src/test/java/org/apache/flume/event/TestEventBuilder.java b/flume-client/src/test/java/org/apache/flume/event/TestEventBuilder.java
similarity index 95%
rename from flume-ng-sdk/src/test/java/org/apache/flume/event/TestEventBuilder.java
rename to flume-client/src/test/java/org/apache/flume/event/TestEventBuilder.java
index 2c5f6b1e..9eb7af96 100644
--- a/flume-ng-sdk/src/test/java/org/apache/flume/event/TestEventBuilder.java
+++ b/flume-client/src/test/java/org/apache/flume/event/TestEventBuilder.java
@@ -18,8 +18,6 @@ package org.apache.flume.event;
import java.util.HashMap;
import java.util.Map;
-import org.apache.flume.Event;
-import org.apache.flume.FlumeException;
import org.junit.Assert;
import org.junit.Test;
@@ -51,7 +49,7 @@ public class TestEventBuilder {
Assert.assertEquals("e1 has a one key", "1", e1.getHeaders().get("one"));
}
- @Test(expected = FlumeException.class)
+ @Test(expected = IllegalArgumentException.class)
public void testJsonEventUnsupportedEncoding() {
JSONEvent jsonEvent = new JSONEvent();
jsonEvent.setCharset("dummy");
diff --git a/flume-ng-sdk/src/test/java/org/apache/flume/util/AbstractSSLUtilListTest.java b/flume-client/src/test/java/org/apache/flume/util/AbstractSSLUtilListTest.java
similarity index 100%
rename from flume-ng-sdk/src/test/java/org/apache/flume/util/AbstractSSLUtilListTest.java
rename to flume-client/src/test/java/org/apache/flume/util/AbstractSSLUtilListTest.java
diff --git a/flume-ng-sdk/src/test/java/org/apache/flume/util/AbstractSSLUtilTest.java b/flume-client/src/test/java/org/apache/flume/util/AbstractSSLUtilTest.java
similarity index 100%
rename from flume-ng-sdk/src/test/java/org/apache/flume/util/AbstractSSLUtilTest.java
rename to flume-client/src/test/java/org/apache/flume/util/AbstractSSLUtilTest.java
diff --git a/flume-ng-sdk/src/test/java/org/apache/flume/util/SSLUtilExcludeCipherSuitesTest.java b/flume-client/src/test/java/org/apache/flume/util/SSLUtilExcludeCipherSuitesTest.java
similarity index 100%
rename from flume-ng-sdk/src/test/java/org/apache/flume/util/SSLUtilExcludeCipherSuitesTest.java
rename to flume-client/src/test/java/org/apache/flume/util/SSLUtilExcludeCipherSuitesTest.java
diff --git a/flume-ng-sdk/src/test/java/org/apache/flume/util/SSLUtilExcludeProtocolsTest.java b/flume-client/src/test/java/org/apache/flume/util/SSLUtilExcludeProtocolsTest.java
similarity index 100%
rename from flume-ng-sdk/src/test/java/org/apache/flume/util/SSLUtilExcludeProtocolsTest.java
rename to flume-client/src/test/java/org/apache/flume/util/SSLUtilExcludeProtocolsTest.java
diff --git a/flume-ng-sdk/src/test/java/org/apache/flume/util/SSLUtilIncludeCipherSuitesTest.java b/flume-client/src/test/java/org/apache/flume/util/SSLUtilIncludeCipherSuitesTest.java
similarity index 100%
rename from flume-ng-sdk/src/test/java/org/apache/flume/util/SSLUtilIncludeCipherSuitesTest.java
rename to flume-client/src/test/java/org/apache/flume/util/SSLUtilIncludeCipherSuitesTest.java
diff --git a/flume-ng-sdk/src/test/java/org/apache/flume/util/SSLUtilIncludeProtocolsTest.java b/flume-client/src/test/java/org/apache/flume/util/SSLUtilIncludeProtocolsTest.java
similarity index 100%
rename from flume-ng-sdk/src/test/java/org/apache/flume/util/SSLUtilIncludeProtocolsTest.java
rename to flume-client/src/test/java/org/apache/flume/util/SSLUtilIncludeProtocolsTest.java
diff --git a/flume-ng-sdk/src/test/java/org/apache/flume/util/SSLUtilKeystorePasswordTest.java b/flume-client/src/test/java/org/apache/flume/util/SSLUtilKeystorePasswordTest.java
similarity index 100%
rename from flume-ng-sdk/src/test/java/org/apache/flume/util/SSLUtilKeystorePasswordTest.java
rename to flume-client/src/test/java/org/apache/flume/util/SSLUtilKeystorePasswordTest.java
diff --git a/flume-ng-sdk/src/test/java/org/apache/flume/util/SSLUtilKeystorePathTest.java b/flume-client/src/test/java/org/apache/flume/util/SSLUtilKeystorePathTest.java
similarity index 100%
rename from flume-ng-sdk/src/test/java/org/apache/flume/util/SSLUtilKeystorePathTest.java
rename to flume-client/src/test/java/org/apache/flume/util/SSLUtilKeystorePathTest.java
diff --git a/flume-ng-sdk/src/test/java/org/apache/flume/util/SSLUtilKeystoreTypeTest.java b/flume-client/src/test/java/org/apache/flume/util/SSLUtilKeystoreTypeTest.java
similarity index 100%
rename from flume-ng-sdk/src/test/java/org/apache/flume/util/SSLUtilKeystoreTypeTest.java
rename to flume-client/src/test/java/org/apache/flume/util/SSLUtilKeystoreTypeTest.java
diff --git a/flume-ng-sdk/src/test/java/org/apache/flume/util/SSLUtilKeystoreTypeWithDefaultTest.java b/flume-client/src/test/java/org/apache/flume/util/SSLUtilKeystoreTypeWithDefaultTest.java
similarity index 100%
rename from flume-ng-sdk/src/test/java/org/apache/flume/util/SSLUtilKeystoreTypeWithDefaultTest.java
rename to flume-client/src/test/java/org/apache/flume/util/SSLUtilKeystoreTypeWithDefaultTest.java
diff --git a/flume-ng-sdk/src/test/java/org/apache/flume/util/SSLUtilTruststorePasswordTest.java b/flume-client/src/test/java/org/apache/flume/util/SSLUtilTruststorePasswordTest.java
similarity index 100%
rename from flume-ng-sdk/src/test/java/org/apache/flume/util/SSLUtilTruststorePasswordTest.java
rename to flume-client/src/test/java/org/apache/flume/util/SSLUtilTruststorePasswordTest.java
diff --git a/flume-ng-sdk/src/test/java/org/apache/flume/util/SSLUtilTruststorePathTest.java b/flume-client/src/test/java/org/apache/flume/util/SSLUtilTruststorePathTest.java
similarity index 100%
rename from flume-ng-sdk/src/test/java/org/apache/flume/util/SSLUtilTruststorePathTest.java
rename to flume-client/src/test/java/org/apache/flume/util/SSLUtilTruststorePathTest.java
diff --git a/flume-ng-sdk/src/test/java/org/apache/flume/util/SSLUtilTruststoreTypeTest.java b/flume-client/src/test/java/org/apache/flume/util/SSLUtilTruststoreTypeTest.java
similarity index 100%
rename from flume-ng-sdk/src/test/java/org/apache/flume/util/SSLUtilTruststoreTypeTest.java
rename to flume-client/src/test/java/org/apache/flume/util/SSLUtilTruststoreTypeTest.java
diff --git a/flume-ng-sdk/src/test/java/org/apache/flume/util/SSLUtilTruststoreTypeWithDefaultTest.java b/flume-client/src/test/java/org/apache/flume/util/SSLUtilTruststoreTypeWithDefaultTest.java
similarity index 100%
rename from flume-ng-sdk/src/test/java/org/apache/flume/util/SSLUtilTruststoreTypeWithDefaultTest.java
rename to flume-client/src/test/java/org/apache/flume/util/SSLUtilTruststoreTypeWithDefaultTest.java
diff --git a/flume-client/src/test/resources/log4j2.xml b/flume-client/src/test/resources/log4j2.xml
new file mode 100644
index 00000000..39d679ad
--- /dev/null
+++ b/flume-client/src/test/resources/log4j2.xml
@@ -0,0 +1,30 @@
+<?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="TRACE"/>
+ <Root level="TRACE">
+ <AppenderRef ref="Console" />
+ </Root>
+ </Loggers>
+</Configuration>
diff --git a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/EventUtils.java b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/EventUtils.java
index 11f52389..e58bca55 100644
--- a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/EventUtils.java
+++ b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/EventUtils.java
@@ -16,7 +16,7 @@
*/
package org.apache.flume.channel.file;
-import org.apache.flume.Event;
+import org.apache.flume.event.Event;
/**
*
diff --git a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/FileChannel.java b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/FileChannel.java
index 2c7d52d2..c26a1db0 100644
--- a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/FileChannel.java
+++ b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/FileChannel.java
@@ -30,7 +30,6 @@ import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import org.apache.flume.Channel;
import org.apache.flume.Context;
-import org.apache.flume.Event;
import org.apache.flume.annotations.Disposable;
import org.apache.flume.annotations.InterfaceAudience;
import org.apache.flume.annotations.InterfaceStability;
@@ -42,6 +41,7 @@ import org.apache.flume.channel.file.encryption.KeyProvider;
import org.apache.flume.channel.file.encryption.KeyProviderFactory;
import org.apache.flume.channel.file.instrumentation.FileChannelCounter;
import org.apache.flume.conf.TransactionCapacitySupported;
+import org.apache.flume.event.Event;
import org.apache.flume.exception.ChannelException;
import org.apache.flume.exception.ChannelFullException;
import org.apache.logging.log4j.LogManager;
diff --git a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/FlumeEvent.java b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/FlumeEvent.java
index df2d9462..3cd748be 100644
--- a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/FlumeEvent.java
+++ b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/FlumeEvent.java
@@ -27,7 +27,7 @@ import java.nio.charset.CharsetEncoder;
import java.nio.charset.CodingErrorAction;
import java.util.HashMap;
import java.util.Map;
-import org.apache.flume.Event;
+import org.apache.flume.event.Event;
/**
* Persistable wrapper for Event
diff --git a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/Log.java b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/Log.java
index c7da2159..beed70f5 100644
--- a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/Log.java
+++ b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/Log.java
@@ -49,11 +49,11 @@ import java.util.concurrent.locks.ReentrantReadWriteLock.ReadLock;
import java.util.concurrent.locks.ReentrantReadWriteLock.WriteLock;
import javax.annotation.Nullable;
import org.apache.commons.io.FileUtils;
-import org.apache.flume.Event;
import org.apache.flume.annotations.InterfaceAudience;
import org.apache.flume.annotations.InterfaceStability;
import org.apache.flume.channel.file.encryption.KeyProvider;
import org.apache.flume.channel.file.instrumentation.FileChannelCounter;
+import org.apache.flume.event.Event;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
diff --git a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/instrumentation/FileChannelCounterMBean.java b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/instrumentation/FileChannelCounterMBean.java
index 4e699d12..39b24452 100644
--- a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/instrumentation/FileChannelCounterMBean.java
+++ b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/instrumentation/FileChannelCounterMBean.java
@@ -43,7 +43,7 @@ public interface FileChannelCounterMBean extends ChannelCounterMBean {
/**
* A count of the number of IOExceptions encountered while trying to put() onto the channel.
* @see org.apache.flume.channel.file.FileChannel.FileBackedTransaction
- * #doPut(org.apache.flume.Event)
+ * #doPut(org.apache.flume.event.Event)
*/
long getEventPutErrorCount();
diff --git a/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestEventUtils.java b/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestEventUtils.java
index 1324ec63..be8282ff 100644
--- a/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestEventUtils.java
+++ b/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestEventUtils.java
@@ -17,7 +17,7 @@
package org.apache.flume.channel.file;
import junit.framework.Assert;
-import org.apache.flume.Event;
+import org.apache.flume.event.Event;
import org.junit.Test;
public class TestEventUtils {
diff --git a/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestFileChannel.java b/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestFileChannel.java
index e9444e4d..3ed49ddb 100644
--- a/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestFileChannel.java
+++ b/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestFileChannel.java
@@ -48,12 +48,12 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
-import org.apache.flume.Event;
import org.apache.flume.Transaction;
import org.apache.flume.channel.file.FileChannel.FileBackedTransaction;
import org.apache.flume.channel.file.FlumeEventQueue.InflightEventWrapper;
import org.apache.flume.channel.file.instrumentation.FileChannelCounter;
import org.apache.flume.conf.Configurables;
+import org.apache.flume.event.Event;
import org.apache.flume.event.EventBuilder;
import org.apache.flume.exception.ChannelException;
import org.apache.logging.log4j.LogManager;
diff --git a/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestUtils.java b/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestUtils.java
index 2fc4aa93..9707cc37 100644
--- a/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestUtils.java
+++ b/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestUtils.java
@@ -44,9 +44,9 @@ import java.util.zip.GZIPInputStream;
import org.apache.commons.io.FileUtils;
import org.apache.flume.Channel;
import org.apache.flume.Context;
-import org.apache.flume.Event;
import org.apache.flume.Transaction;
import org.apache.flume.conf.Configurables;
+import org.apache.flume.event.Event;
import org.apache.flume.event.EventBuilder;
import org.apache.flume.exception.ChannelException;
import org.junit.Assert;
diff --git a/flume-ng-channels/flume-spillable-memory-channel/src/main/java/org/apache/flume/channel/SpillableMemoryChannel.java b/flume-ng-channels/flume-spillable-memory-channel/src/main/java/org/apache/flume/channel/SpillableMemoryChannel.java
index 25c8ada3..fe19a9d5 100644
--- a/flume-ng-channels/flume-spillable-memory-channel/src/main/java/org/apache/flume/channel/SpillableMemoryChannel.java
+++ b/flume-ng-channels/flume-spillable-memory-channel/src/main/java/org/apache/flume/channel/SpillableMemoryChannel.java
@@ -23,12 +23,12 @@ import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import javax.annotation.concurrent.GuardedBy;
import org.apache.flume.Context;
-import org.apache.flume.Event;
import org.apache.flume.Transaction;
import org.apache.flume.annotations.InterfaceAudience;
import org.apache.flume.annotations.InterfaceStability;
import org.apache.flume.annotations.Recyclable;
import org.apache.flume.channel.file.FileChannel;
+import org.apache.flume.event.Event;
import org.apache.flume.exception.ChannelException;
import org.apache.flume.exception.ChannelFullException;
import org.apache.flume.instrumentation.ChannelCounter;
diff --git a/flume-ng-channels/flume-spillable-memory-channel/src/test/java/org/apache/flume/channel/TestSpillableMemoryChannel.java b/flume-ng-channels/flume-spillable-memory-channel/src/test/java/org/apache/flume/channel/TestSpillableMemoryChannel.java
index 2c993a77..f9effb0c 100644
--- a/flume-ng-channels/flume-spillable-memory-channel/src/test/java/org/apache/flume/channel/TestSpillableMemoryChannel.java
+++ b/flume-ng-channels/flume-spillable-memory-channel/src/test/java/org/apache/flume/channel/TestSpillableMemoryChannel.java
@@ -26,10 +26,10 @@ import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import org.apache.flume.Context;
-import org.apache.flume.Event;
import org.apache.flume.Transaction;
import org.apache.flume.channel.file.FileChannelConfiguration;
import org.apache.flume.conf.Configurables;
+import org.apache.flume.event.Event;
import org.apache.flume.event.EventBuilder;
import org.apache.flume.exception.ChannelException;
import org.apache.flume.exception.ChannelFullException;
diff --git a/flume-ng-core/src/main/java/org/apache/flume/Channel.java b/flume-ng-core/src/main/java/org/apache/flume/Channel.java
index e3366dc2..f54c9544 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/Channel.java
+++ b/flume-ng-core/src/main/java/org/apache/flume/Channel.java
@@ -18,6 +18,7 @@ package org.apache.flume;
import org.apache.flume.annotations.InterfaceAudience;
import org.apache.flume.annotations.InterfaceStability;
+import org.apache.flume.event.Event;
import org.apache.flume.exception.ChannelException;
import org.apache.flume.lifecycle.LifecycleAware;
diff --git a/flume-ng-core/src/main/java/org/apache/flume/ChannelSelector.java b/flume-ng-core/src/main/java/org/apache/flume/ChannelSelector.java
index 2b59ff89..9fbd2e06 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/ChannelSelector.java
+++ b/flume-ng-core/src/main/java/org/apache/flume/ChannelSelector.java
@@ -18,6 +18,7 @@ package org.apache.flume;
import java.util.List;
import org.apache.flume.conf.Configurable;
+import org.apache.flume.event.Event;
/**
* <p>
diff --git a/flume-ng-core/src/main/java/org/apache/flume/channel/BasicChannelSemantics.java b/flume-ng-core/src/main/java/org/apache/flume/channel/BasicChannelSemantics.java
index 1c7e1fed..c6f77b16 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/channel/BasicChannelSemantics.java
+++ b/flume-ng-core/src/main/java/org/apache/flume/channel/BasicChannelSemantics.java
@@ -18,10 +18,10 @@ package org.apache.flume.channel;
import com.google.common.base.Preconditions;
import org.apache.flume.Channel;
-import org.apache.flume.Event;
import org.apache.flume.Transaction;
import org.apache.flume.annotations.InterfaceAudience;
import org.apache.flume.annotations.InterfaceStability;
+import org.apache.flume.event.Event;
import org.apache.flume.exception.ChannelException;
/**
diff --git a/flume-ng-core/src/main/java/org/apache/flume/channel/BasicTransactionSemantics.java b/flume-ng-core/src/main/java/org/apache/flume/channel/BasicTransactionSemantics.java
index f1918b37..a5b396cb 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/channel/BasicTransactionSemantics.java
+++ b/flume-ng-core/src/main/java/org/apache/flume/channel/BasicTransactionSemantics.java
@@ -18,8 +18,8 @@ package org.apache.flume.channel;
import com.google.common.base.Preconditions;
import org.apache.flume.Channel;
-import org.apache.flume.Event;
import org.apache.flume.Transaction;
+import org.apache.flume.event.Event;
import org.apache.flume.exception.ChannelException;
/**
diff --git a/flume-ng-core/src/main/java/org/apache/flume/channel/ChannelProcessor.java b/flume-ng-core/src/main/java/org/apache/flume/channel/ChannelProcessor.java
index 30ef43b5..aae8c00f 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/channel/ChannelProcessor.java
+++ b/flume-ng-core/src/main/java/org/apache/flume/channel/ChannelProcessor.java
@@ -25,10 +25,10 @@ import java.util.Map;
import org.apache.flume.Channel;
import org.apache.flume.ChannelSelector;
import org.apache.flume.Context;
-import org.apache.flume.Event;
import org.apache.flume.FlumeException;
import org.apache.flume.Transaction;
import org.apache.flume.conf.Configurable;
+import org.apache.flume.event.Event;
import org.apache.flume.exception.ChannelException;
import org.apache.flume.interceptor.Interceptor;
import org.apache.flume.interceptor.InterceptorBuilderFactory;
diff --git a/flume-ng-core/src/main/java/org/apache/flume/channel/ChannelUtils.java b/flume-ng-core/src/main/java/org/apache/flume/channel/ChannelUtils.java
index 2d261a85..515e93e1 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/channel/ChannelUtils.java
+++ b/flume-ng-core/src/main/java/org/apache/flume/channel/ChannelUtils.java
@@ -22,8 +22,8 @@ import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.Executors;
import org.apache.flume.Channel;
-import org.apache.flume.Event;
import org.apache.flume.Transaction;
+import org.apache.flume.event.Event;
import org.apache.flume.exception.ChannelException;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
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
index 78e656a2..dd4cb2cb 100644
--- 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
@@ -24,8 +24,8 @@ 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 org.apache.flume.conf.internal.SuppressFBWarnings;
+import org.apache.flume.event.Event;
/**
* Load balancing channel selector. This selector allows for load balancing
diff --git a/flume-ng-core/src/main/java/org/apache/flume/channel/MemoryChannel.java b/flume-ng-core/src/main/java/org/apache/flume/channel/MemoryChannel.java
index b0c6c3c6..498b54b9 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/channel/MemoryChannel.java
+++ b/flume-ng-core/src/main/java/org/apache/flume/channel/MemoryChannel.java
@@ -23,11 +23,11 @@ import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import javax.annotation.concurrent.GuardedBy;
import org.apache.flume.Context;
-import org.apache.flume.Event;
import org.apache.flume.annotations.InterfaceAudience;
import org.apache.flume.annotations.InterfaceStability;
import org.apache.flume.annotations.Recyclable;
import org.apache.flume.conf.TransactionCapacitySupported;
+import org.apache.flume.event.Event;
import org.apache.flume.exception.ChannelException;
import org.apache.flume.exception.ChannelFullException;
import org.apache.flume.instrumentation.ChannelCounter;
diff --git a/flume-ng-core/src/main/java/org/apache/flume/channel/MultiplexingChannelSelector.java b/flume-ng-core/src/main/java/org/apache/flume/channel/MultiplexingChannelSelector.java
index d92d4cb0..1d1fd6ae 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/channel/MultiplexingChannelSelector.java
+++ b/flume-ng-core/src/main/java/org/apache/flume/channel/MultiplexingChannelSelector.java
@@ -22,9 +22,9 @@ import java.util.List;
import java.util.Map;
import org.apache.flume.Channel;
import org.apache.flume.Context;
-import org.apache.flume.Event;
import org.apache.flume.FlumeException;
import org.apache.flume.conf.internal.SuppressFBWarnings;
+import org.apache.flume.event.Event;
@SuppressFBWarnings("UWF_FIELD_NOT_INITIALIZED_IN_CONSTRUCTOR")
public class MultiplexingChannelSelector extends AbstractChannelSelector {
diff --git a/flume-ng-core/src/main/java/org/apache/flume/channel/PseudoTxnMemoryChannel.java b/flume-ng-core/src/main/java/org/apache/flume/channel/PseudoTxnMemoryChannel.java
index 098ba6c8..957e4112 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/channel/PseudoTxnMemoryChannel.java
+++ b/flume-ng-core/src/main/java/org/apache/flume/channel/PseudoTxnMemoryChannel.java
@@ -22,8 +22,8 @@ import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import org.apache.flume.Channel;
import org.apache.flume.Context;
-import org.apache.flume.Event;
import org.apache.flume.Transaction;
+import org.apache.flume.event.Event;
import org.apache.flume.exception.ChannelException;
import org.apache.flume.instrumentation.ChannelCounter;
diff --git a/flume-ng-core/src/main/java/org/apache/flume/channel/ReplicatingChannelSelector.java b/flume-ng-core/src/main/java/org/apache/flume/channel/ReplicatingChannelSelector.java
index 8289dbf4..bee2503f 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/channel/ReplicatingChannelSelector.java
+++ b/flume-ng-core/src/main/java/org/apache/flume/channel/ReplicatingChannelSelector.java
@@ -21,7 +21,7 @@ import java.util.List;
import java.util.Map;
import org.apache.flume.Channel;
import org.apache.flume.Context;
-import org.apache.flume.Event;
+import org.apache.flume.event.Event;
/**
* Replicating channel selector. This selector allows the event to be placed
diff --git a/flume-ng-core/src/main/java/org/apache/flume/channel/RoutableProxyChannelSelector.java b/flume-ng-core/src/main/java/org/apache/flume/channel/RoutableProxyChannelSelector.java
index 00369df1..074895a2 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/channel/RoutableProxyChannelSelector.java
+++ b/flume-ng-core/src/main/java/org/apache/flume/channel/RoutableProxyChannelSelector.java
@@ -27,8 +27,8 @@ import org.apache.commons.lang3.StringUtils;
import org.apache.flume.Channel;
import org.apache.flume.ChannelSelector;
import org.apache.flume.Context;
-import org.apache.flume.Event;
import org.apache.flume.conf.Configurables;
+import org.apache.flume.event.Event;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
diff --git a/flume-ng-core/src/main/java/org/apache/flume/event/EventHelper.java b/flume-ng-core/src/main/java/org/apache/flume/core/EventHelper.java
similarity index 97%
rename from flume-ng-core/src/main/java/org/apache/flume/event/EventHelper.java
rename to flume-ng-core/src/main/java/org/apache/flume/core/EventHelper.java
index 733377f2..b2979ac6 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/event/EventHelper.java
+++ b/flume-ng-core/src/main/java/org/apache/flume/core/EventHelper.java
@@ -14,12 +14,12 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.flume.event;
+package org.apache.flume.core;
import java.io.ByteArrayOutputStream;
import java.util.Arrays;
import org.apache.commons.io.HexDump;
-import org.apache.flume.Event;
+import org.apache.flume.event.Event;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
diff --git a/flume-ng-core/src/main/java/org/apache/flume/formatter/output/EventFormatter.java b/flume-ng-core/src/main/java/org/apache/flume/formatter/output/EventFormatter.java
index e8c8900c..d34e4310 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/formatter/output/EventFormatter.java
+++ b/flume-ng-core/src/main/java/org/apache/flume/formatter/output/EventFormatter.java
@@ -16,7 +16,7 @@
*/
package org.apache.flume.formatter.output;
-import org.apache.flume.Event;
+import org.apache.flume.event.Event;
@Deprecated
public interface EventFormatter {
diff --git a/flume-ng-core/src/main/java/org/apache/flume/formatter/output/TextDelimitedOutputFormatter.java b/flume-ng-core/src/main/java/org/apache/flume/formatter/output/TextDelimitedOutputFormatter.java
index 87b08157..06433b5c 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/formatter/output/TextDelimitedOutputFormatter.java
+++ b/flume-ng-core/src/main/java/org/apache/flume/formatter/output/TextDelimitedOutputFormatter.java
@@ -16,7 +16,7 @@
*/
package org.apache.flume.formatter.output;
-import org.apache.flume.Event;
+import org.apache.flume.event.Event;
@Deprecated
public class TextDelimitedOutputFormatter implements EventFormatter {
diff --git a/flume-ng-core/src/main/java/org/apache/flume/interceptor/HostInterceptor.java b/flume-ng-core/src/main/java/org/apache/flume/interceptor/HostInterceptor.java
index 94ea167f..8ee4ac54 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/interceptor/HostInterceptor.java
+++ b/flume-ng-core/src/main/java/org/apache/flume/interceptor/HostInterceptor.java
@@ -28,7 +28,7 @@ import java.net.UnknownHostException;
import java.util.List;
import java.util.Map;
import org.apache.flume.Context;
-import org.apache.flume.Event;
+import org.apache.flume.event.Event;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
diff --git a/flume-ng-core/src/main/java/org/apache/flume/interceptor/Interceptor.java b/flume-ng-core/src/main/java/org/apache/flume/interceptor/Interceptor.java
index 7fc87a70..53c2ba84 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/interceptor/Interceptor.java
+++ b/flume-ng-core/src/main/java/org/apache/flume/interceptor/Interceptor.java
@@ -17,10 +17,10 @@
package org.apache.flume.interceptor;
import java.util.List;
-import org.apache.flume.Event;
import org.apache.flume.annotations.InterfaceAudience;
import org.apache.flume.annotations.InterfaceStability;
import org.apache.flume.conf.Configurable;
+import org.apache.flume.event.Event;
@InterfaceAudience.Public
@InterfaceStability.Stable
diff --git a/flume-ng-core/src/main/java/org/apache/flume/interceptor/InterceptorChain.java b/flume-ng-core/src/main/java/org/apache/flume/interceptor/InterceptorChain.java
index 25914929..977beeba 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/interceptor/InterceptorChain.java
+++ b/flume-ng-core/src/main/java/org/apache/flume/interceptor/InterceptorChain.java
@@ -20,7 +20,7 @@ import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import java.util.Iterator;
import java.util.List;
-import org.apache.flume.Event;
+import org.apache.flume.event.Event;
/**
* Implementation of Interceptor that calls a list of other Interceptors
diff --git a/flume-ng-core/src/main/java/org/apache/flume/interceptor/RegexExtractorInterceptor.java b/flume-ng-core/src/main/java/org/apache/flume/interceptor/RegexExtractorInterceptor.java
index e17183d3..a447f04e 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/interceptor/RegexExtractorInterceptor.java
+++ b/flume-ng-core/src/main/java/org/apache/flume/interceptor/RegexExtractorInterceptor.java
@@ -26,7 +26,7 @@ import java.util.regex.Matcher;
import java.util.regex.Pattern;
import org.apache.commons.lang3.StringUtils;
import org.apache.flume.Context;
-import org.apache.flume.Event;
+import org.apache.flume.event.Event;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
diff --git a/flume-ng-core/src/main/java/org/apache/flume/interceptor/RegexFilteringInterceptor.java b/flume-ng-core/src/main/java/org/apache/flume/interceptor/RegexFilteringInterceptor.java
index d3a2d2ec..ba31aa02 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/interceptor/RegexFilteringInterceptor.java
+++ b/flume-ng-core/src/main/java/org/apache/flume/interceptor/RegexFilteringInterceptor.java
@@ -25,7 +25,7 @@ import com.google.common.collect.Lists;
import java.util.List;
import java.util.regex.Pattern;
import org.apache.flume.Context;
-import org.apache.flume.Event;
+import org.apache.flume.event.Event;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
diff --git a/flume-ng-core/src/main/java/org/apache/flume/interceptor/RemoveHeaderInterceptor.java b/flume-ng-core/src/main/java/org/apache/flume/interceptor/RemoveHeaderInterceptor.java
index 941da03d..041339bb 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/interceptor/RemoveHeaderInterceptor.java
+++ b/flume-ng-core/src/main/java/org/apache/flume/interceptor/RemoveHeaderInterceptor.java
@@ -26,8 +26,8 @@ import java.util.Set;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import org.apache.flume.Context;
-import org.apache.flume.Event;
import org.apache.flume.conf.LogPrivacyUtil;
+import org.apache.flume.event.Event;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@@ -98,7 +98,7 @@ public class RemoveHeaderInterceptor implements Interceptor {
}
/**
- * @see org.apache.flume.interceptor.Interceptor#intercept(org.apache.flume.Event)
+ * @see org.apache.flume.interceptor.Interceptor#intercept(org.apache.flume.event.Event)
*/
@Override
public Event intercept(final Event event) {
diff --git a/flume-ng-core/src/main/java/org/apache/flume/interceptor/SearchAndReplaceInterceptor.java b/flume-ng-core/src/main/java/org/apache/flume/interceptor/SearchAndReplaceInterceptor.java
index 232d4e57..5b087f09 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/interceptor/SearchAndReplaceInterceptor.java
+++ b/flume-ng-core/src/main/java/org/apache/flume/interceptor/SearchAndReplaceInterceptor.java
@@ -24,7 +24,7 @@ import java.util.regex.Matcher;
import java.util.regex.Pattern;
import org.apache.commons.lang3.StringUtils;
import org.apache.flume.Context;
-import org.apache.flume.Event;
+import org.apache.flume.event.Event;
/**
* <p>
diff --git a/flume-ng-core/src/main/java/org/apache/flume/interceptor/StaticInterceptor.java b/flume-ng-core/src/main/java/org/apache/flume/interceptor/StaticInterceptor.java
index e762616d..91e25731 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/interceptor/StaticInterceptor.java
+++ b/flume-ng-core/src/main/java/org/apache/flume/interceptor/StaticInterceptor.java
@@ -19,7 +19,7 @@ package org.apache.flume.interceptor;
import java.util.List;
import java.util.Map;
import org.apache.flume.Context;
-import org.apache.flume.Event;
+import org.apache.flume.event.Event;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
diff --git a/flume-ng-core/src/main/java/org/apache/flume/interceptor/TimestampInterceptor.java b/flume-ng-core/src/main/java/org/apache/flume/interceptor/TimestampInterceptor.java
index 6b558850..233c0d8b 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/interceptor/TimestampInterceptor.java
+++ b/flume-ng-core/src/main/java/org/apache/flume/interceptor/TimestampInterceptor.java
@@ -24,7 +24,7 @@ import static org.apache.flume.interceptor.TimestampInterceptor.Constants.DEFAUL
import java.util.List;
import java.util.Map;
import org.apache.flume.Context;
-import org.apache.flume.Event;
+import org.apache.flume.event.Event;
/**
* Simple Interceptor class that sets the current system timestamp on all events
diff --git a/flume-ng-core/src/main/java/org/apache/flume/serialization/BodyTextEventSerializer.java b/flume-ng-core/src/main/java/org/apache/flume/serialization/BodyTextEventSerializer.java
index 3aefcf4f..911f9794 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/serialization/BodyTextEventSerializer.java
+++ b/flume-ng-core/src/main/java/org/apache/flume/serialization/BodyTextEventSerializer.java
@@ -19,7 +19,7 @@ package org.apache.flume.serialization;
import java.io.IOException;
import java.io.OutputStream;
import org.apache.flume.Context;
-import org.apache.flume.Event;
+import org.apache.flume.event.Event;
/**
* This class simply writes the body of the event to the output stream
diff --git a/flume-ng-core/src/main/java/org/apache/flume/serialization/EventSerializer.java b/flume-ng-core/src/main/java/org/apache/flume/serialization/EventSerializer.java
index ce482217..0dcf70bc 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/serialization/EventSerializer.java
+++ b/flume-ng-core/src/main/java/org/apache/flume/serialization/EventSerializer.java
@@ -19,9 +19,9 @@ package org.apache.flume.serialization;
import java.io.IOException;
import java.io.OutputStream;
import org.apache.flume.Context;
-import org.apache.flume.Event;
import org.apache.flume.annotations.InterfaceAudience;
import org.apache.flume.annotations.InterfaceStability;
+import org.apache.flume.event.Event;
/**
* <p>
diff --git a/flume-ng-core/src/main/java/org/apache/flume/serialization/HeaderAndBodyTextEventSerializer.java b/flume-ng-core/src/main/java/org/apache/flume/serialization/HeaderAndBodyTextEventSerializer.java
index 8d92612b..1a0dda42 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/serialization/HeaderAndBodyTextEventSerializer.java
+++ b/flume-ng-core/src/main/java/org/apache/flume/serialization/HeaderAndBodyTextEventSerializer.java
@@ -19,7 +19,7 @@ package org.apache.flume.serialization;
import java.io.IOException;
import java.io.OutputStream;
import org.apache.flume.Context;
-import org.apache.flume.Event;
+import org.apache.flume.event.Event;
/**
* This class simply writes the body of the event to the output stream
diff --git a/flume-ng-core/src/main/java/org/apache/flume/sink/AbstractRpcSink.java b/flume-ng-core/src/main/java/org/apache/flume/sink/AbstractRpcSink.java
index 9dc4844b..a0c49797 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/sink/AbstractRpcSink.java
+++ b/flume-ng-core/src/main/java/org/apache/flume/sink/AbstractRpcSink.java
@@ -20,6 +20,7 @@ import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import java.io.IOException;
import java.util.List;
import java.util.Map.Entry;
import java.util.Properties;
@@ -29,7 +30,6 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.flume.Channel;
import org.apache.flume.Context;
-import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.FlumeException;
import org.apache.flume.Transaction;
@@ -38,6 +38,7 @@ import org.apache.flume.api.RpcClient;
import org.apache.flume.api.RpcClientConfigurationConstants;
import org.apache.flume.conf.BatchSizeSupported;
import org.apache.flume.conf.Configurable;
+import org.apache.flume.event.Event;
import org.apache.flume.exception.ChannelException;
import org.apache.flume.instrumentation.SinkCounter;
import org.apache.logging.log4j.LogManager;
@@ -190,8 +191,9 @@ public abstract class AbstractRpcSink extends AbstractSink implements Configurab
* connection needs to be created to the next hop.
* @param props
* @return
+ * @throws IOException if the client fails to connect.
*/
- protected abstract RpcClient initializeRpcClient(Properties props);
+ protected abstract RpcClient initializeRpcClient(Properties props) throws IOException;
/**
* If this function is called successively without calling
@@ -249,7 +251,7 @@ public abstract class AbstractRpcSink extends AbstractSink implements Configurab
try {
client.close();
sinkCounter.incrementConnectionClosedCount();
- } catch (FlumeException e) {
+ } catch (IOException e) {
sinkCounter.incrementConnectionFailedCount();
logger.error(
"Rpc sink " + getName() + ": Attempt to close Rpc " + "client failed. Exception follows.", e);
diff --git a/flume-ng-core/src/main/java/org/apache/flume/sink/LoggerSink.java b/flume-ng-core/src/main/java/org/apache/flume/sink/LoggerSink.java
index fb1f4166..939dfb26 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/sink/LoggerSink.java
+++ b/flume-ng-core/src/main/java/org/apache/flume/sink/LoggerSink.java
@@ -19,11 +19,11 @@ package org.apache.flume.sink;
import com.google.common.base.Strings;
import org.apache.flume.Channel;
import org.apache.flume.Context;
-import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.Transaction;
import org.apache.flume.conf.Configurable;
-import org.apache.flume.event.EventHelper;
+import org.apache.flume.core.EventHelper;
+import org.apache.flume.event.Event;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
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 51290b04..4c8470f4 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
@@ -20,12 +20,12 @@ import com.google.common.base.Preconditions;
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.Transaction;
import org.apache.flume.conf.BatchSizeSupported;
import org.apache.flume.conf.Configurable;
+import org.apache.flume.event.Event;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
diff --git a/flume-ng-core/src/main/java/org/apache/flume/sink/RollingFileSink.java b/flume-ng-core/src/main/java/org/apache/flume/sink/RollingFileSink.java
index 559a7468..68749068 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/sink/RollingFileSink.java
+++ b/flume-ng-core/src/main/java/org/apache/flume/sink/RollingFileSink.java
@@ -28,11 +28,11 @@ import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import org.apache.flume.Channel;
import org.apache.flume.Context;
-import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.Transaction;
import org.apache.flume.conf.BatchSizeSupported;
import org.apache.flume.conf.Configurable;
+import org.apache.flume.event.Event;
import org.apache.flume.formatter.output.PathManager;
import org.apache.flume.formatter.output.PathManagerFactory;
import org.apache.flume.instrumentation.SinkCounter;
diff --git a/flume-ng-core/src/main/java/org/apache/flume/source/ExecSource.java b/flume-ng-core/src/main/java/org/apache/flume/source/ExecSource.java
index 17b03c58..3375d664 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/source/ExecSource.java
+++ b/flume-ng-core/src/main/java/org/apache/flume/source/ExecSource.java
@@ -33,7 +33,6 @@ import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import org.apache.flume.Channel;
import org.apache.flume.Context;
-import org.apache.flume.Event;
import org.apache.flume.EventDrivenSource;
import org.apache.flume.Source;
import org.apache.flume.SystemClock;
@@ -41,6 +40,7 @@ import org.apache.flume.channel.ChannelProcessor;
import org.apache.flume.conf.BatchSizeSupported;
import org.apache.flume.conf.Configurable;
import org.apache.flume.conf.internal.SuppressFBWarnings;
+import org.apache.flume.event.Event;
import org.apache.flume.event.EventBuilder;
import org.apache.flume.instrumentation.SourceCounter;
import org.apache.logging.log4j.LogManager;
diff --git a/flume-ng-core/src/main/java/org/apache/flume/source/NetcatSource.java b/flume-ng-core/src/main/java/org/apache/flume/source/NetcatSource.java
index 03fdb7f8..0e81586c 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/source/NetcatSource.java
+++ b/flume-ng-core/src/main/java/org/apache/flume/source/NetcatSource.java
@@ -35,12 +35,12 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.flume.Context;
import org.apache.flume.CounterGroup;
-import org.apache.flume.Event;
import org.apache.flume.EventDrivenSource;
import org.apache.flume.FlumeException;
import org.apache.flume.Source;
import org.apache.flume.conf.Configurable;
import org.apache.flume.conf.Configurables;
+import org.apache.flume.event.Event;
import org.apache.flume.event.EventBuilder;
import org.apache.flume.exception.ChannelException;
import org.apache.logging.log4j.LogManager;
diff --git a/flume-ng-core/src/main/java/org/apache/flume/source/SequenceGeneratorSource.java b/flume-ng-core/src/main/java/org/apache/flume/source/SequenceGeneratorSource.java
index 2d9e1c2a..fbd5b9d7 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/source/SequenceGeneratorSource.java
+++ b/flume-ng-core/src/main/java/org/apache/flume/source/SequenceGeneratorSource.java
@@ -20,11 +20,11 @@ import com.google.common.base.Preconditions;
import java.util.ArrayList;
import java.util.List;
import org.apache.flume.Context;
-import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.FlumeException;
import org.apache.flume.conf.BatchSizeSupported;
import org.apache.flume.conf.Configurable;
+import org.apache.flume.event.Event;
import org.apache.flume.event.EventBuilder;
import org.apache.flume.exception.ChannelException;
import org.apache.flume.instrumentation.SourceCounter;
diff --git a/flume-ng-core/src/main/java/org/apache/flume/source/StressSource.java b/flume-ng-core/src/main/java/org/apache/flume/source/StressSource.java
index 468e57ef..edb1d582 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/source/StressSource.java
+++ b/flume-ng-core/src/main/java/org/apache/flume/source/StressSource.java
@@ -21,11 +21,11 @@ import java.util.Arrays;
import java.util.List;
import org.apache.flume.Context;
import org.apache.flume.CounterGroup;
-import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.FlumeException;
import org.apache.flume.conf.BatchSizeSupported;
import org.apache.flume.conf.Configurable;
+import org.apache.flume.event.Event;
import org.apache.flume.event.EventBuilder;
import org.apache.flume.exception.ChannelException;
import org.apache.flume.source.shaded.guava.RateLimiter;
diff --git a/flume-ng-core/src/test/java/org/apache/flume/channel/AbstractBasicChannelSemanticsTest.java b/flume-ng-core/src/test/java/org/apache/flume/channel/AbstractBasicChannelSemanticsTest.java
index 76912cc1..2de05689 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/channel/AbstractBasicChannelSemanticsTest.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/channel/AbstractBasicChannelSemanticsTest.java
@@ -25,7 +25,7 @@ import java.util.Queue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
-import org.apache.flume.Event;
+import org.apache.flume.event.Event;
import org.apache.flume.event.EventBuilder;
import org.apache.flume.exception.ChannelException;
import org.junit.After;
diff --git a/flume-ng-core/src/test/java/org/apache/flume/channel/MockChannel.java b/flume-ng-core/src/test/java/org/apache/flume/channel/MockChannel.java
index 3b869333..ab21d407 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/channel/MockChannel.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/channel/MockChannel.java
@@ -19,8 +19,8 @@ package org.apache.flume.channel;
import java.util.ArrayList;
import java.util.List;
import org.apache.flume.Channel;
-import org.apache.flume.Event;
import org.apache.flume.Transaction;
+import org.apache.flume.event.Event;
import org.apache.flume.exception.ChannelException;
public class MockChannel extends AbstractChannel {
diff --git a/flume-ng-core/src/test/java/org/apache/flume/channel/MockEvent.java b/flume-ng-core/src/test/java/org/apache/flume/channel/MockEvent.java
index 83b04c83..638b5b45 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/channel/MockEvent.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/channel/MockEvent.java
@@ -18,7 +18,7 @@ package org.apache.flume.channel;
import java.util.HashMap;
import java.util.Map;
-import org.apache.flume.Event;
+import org.apache.flume.event.Event;
public class MockEvent implements Event {
diff --git a/flume-ng-core/src/test/java/org/apache/flume/channel/TestBasicChannelSemantics.java b/flume-ng-core/src/test/java/org/apache/flume/channel/TestBasicChannelSemantics.java
index 3f88b7c6..97a19c2c 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/channel/TestBasicChannelSemantics.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/channel/TestBasicChannelSemantics.java
@@ -17,8 +17,8 @@
package org.apache.flume.channel;
import java.util.concurrent.Future;
-import org.apache.flume.Event;
import org.apache.flume.Transaction;
+import org.apache.flume.event.Event;
import org.junit.Assert;
import org.junit.Test;
diff --git a/flume-ng-core/src/test/java/org/apache/flume/channel/TestChannelProcessor.java b/flume-ng-core/src/test/java/org/apache/flume/channel/TestChannelProcessor.java
index 2b5dd97f..4999e91c 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/channel/TestChannelProcessor.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/channel/TestChannelProcessor.java
@@ -26,9 +26,9 @@ import java.util.List;
import org.apache.flume.Channel;
import org.apache.flume.ChannelSelector;
import org.apache.flume.Context;
-import org.apache.flume.Event;
import org.apache.flume.Transaction;
import org.apache.flume.conf.Configurables;
+import org.apache.flume.event.Event;
import org.apache.flume.event.EventBuilder;
import org.apache.flume.exception.ChannelException;
import org.junit.Assert;
diff --git a/flume-ng-core/src/test/java/org/apache/flume/channel/TestChannelUtils.java b/flume-ng-core/src/test/java/org/apache/flume/channel/TestChannelUtils.java
index 10e07a62..13502c60 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/channel/TestChannelUtils.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/channel/TestChannelUtils.java
@@ -17,7 +17,7 @@
package org.apache.flume.channel;
import java.util.List;
-import org.apache.flume.Event;
+import org.apache.flume.event.Event;
import org.apache.flume.exception.ChannelException;
import org.junit.Assert;
import org.junit.Test;
diff --git a/flume-ng-core/src/test/java/org/apache/flume/channel/TestMemoryChannel.java b/flume-ng-core/src/test/java/org/apache/flume/channel/TestMemoryChannel.java
index b71619d1..31ad41ae 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/channel/TestMemoryChannel.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/channel/TestMemoryChannel.java
@@ -25,10 +25,10 @@ import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.LinkedBlockingDeque;
import org.apache.flume.Context;
-import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.Transaction;
import org.apache.flume.conf.Configurables;
+import org.apache.flume.event.Event;
import org.apache.flume.event.EventBuilder;
import org.apache.flume.event.SimpleEvent;
import org.apache.flume.exception.ChannelException;
diff --git a/flume-ng-core/src/test/java/org/apache/flume/channel/TestMemoryChannelConcurrency.java b/flume-ng-core/src/test/java/org/apache/flume/channel/TestMemoryChannelConcurrency.java
index 8ee407a7..6daa4ac0 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/channel/TestMemoryChannelConcurrency.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/channel/TestMemoryChannelConcurrency.java
@@ -26,9 +26,9 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.flume.Channel;
import org.apache.flume.Context;
-import org.apache.flume.Event;
import org.apache.flume.Transaction;
import org.apache.flume.conf.Configurables;
+import org.apache.flume.event.Event;
import org.apache.flume.event.EventBuilder;
import org.apache.flume.exception.ChannelException;
import org.junit.Assert;
diff --git a/flume-ng-core/src/test/java/org/apache/flume/channel/TestMemoryChannelTransaction.java b/flume-ng-core/src/test/java/org/apache/flume/channel/TestMemoryChannelTransaction.java
index 138e38c0..2484267b 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/channel/TestMemoryChannelTransaction.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/channel/TestMemoryChannelTransaction.java
@@ -18,10 +18,10 @@ package org.apache.flume.channel;
import org.apache.flume.Channel;
import org.apache.flume.Context;
-import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.Transaction;
import org.apache.flume.conf.Configurables;
+import org.apache.flume.event.Event;
import org.apache.flume.event.EventBuilder;
import org.junit.Assert;
import org.junit.Before;
diff --git a/flume-ng-core/src/test/java/org/apache/flume/channel/TestMultiplexingChannelSelector.java b/flume-ng-core/src/test/java/org/apache/flume/channel/TestMultiplexingChannelSelector.java
index 51dd8ff0..b3835fef 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/channel/TestMultiplexingChannelSelector.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/channel/TestMultiplexingChannelSelector.java
@@ -23,7 +23,7 @@ import java.util.Map;
import junit.framework.Assert;
import org.apache.flume.Channel;
import org.apache.flume.ChannelSelector;
-import org.apache.flume.Event;
+import org.apache.flume.event.Event;
import org.junit.Before;
import org.junit.Test;
diff --git a/flume-ng-core/src/test/java/org/apache/flume/channel/TestRoutableProxyChannelSelector.java b/flume-ng-core/src/test/java/org/apache/flume/channel/TestRoutableProxyChannelSelector.java
index f1df71cb..48cc5f27 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/channel/TestRoutableProxyChannelSelector.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/channel/TestRoutableProxyChannelSelector.java
@@ -25,7 +25,7 @@ import java.util.stream.Collectors;
import junit.framework.Assert;
import org.apache.flume.Channel;
import org.apache.flume.ChannelSelector;
-import org.apache.flume.Event;
+import org.apache.flume.event.Event;
import org.apache.flume.event.SimpleEvent;
import org.junit.Test;
diff --git a/flume-ng-core/src/test/java/org/apache/flume/event/TestEventHelper.java b/flume-ng-core/src/test/java/org/apache/flume/core/TestEventHelper.java
similarity index 95%
rename from flume-ng-core/src/test/java/org/apache/flume/event/TestEventHelper.java
rename to flume-ng-core/src/test/java/org/apache/flume/core/TestEventHelper.java
index ae996bce..8e89696f 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/event/TestEventHelper.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/core/TestEventHelper.java
@@ -14,8 +14,9 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.flume.event;
+package org.apache.flume.core;
+import org.apache.flume.event.SimpleEvent;
import org.junit.Assert;
import org.junit.Test;
diff --git a/flume-ng-core/src/test/java/org/apache/flume/interceptor/CensoringInterceptor.java b/flume-ng-core/src/test/java/org/apache/flume/interceptor/CensoringInterceptor.java
index 76d73c08..39900195 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/interceptor/CensoringInterceptor.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/interceptor/CensoringInterceptor.java
@@ -19,7 +19,7 @@ package org.apache.flume.interceptor;
import java.util.List;
import java.util.Map;
import org.apache.flume.Context;
-import org.apache.flume.Event;
+import org.apache.flume.event.Event;
/**
* Simple interceptor that removes a header called "Bad-Words" from all events.
diff --git a/flume-ng-core/src/test/java/org/apache/flume/interceptor/RemoveHeaderInterceptorTest.java b/flume-ng-core/src/test/java/org/apache/flume/interceptor/RemoveHeaderInterceptorTest.java
index 2d23df3f..58d815cb 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/interceptor/RemoveHeaderInterceptorTest.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/interceptor/RemoveHeaderInterceptorTest.java
@@ -22,7 +22,7 @@ import java.util.Map;
import java.util.regex.PatternSyntaxException;
import junit.framework.Assert;
import org.apache.flume.Context;
-import org.apache.flume.Event;
+import org.apache.flume.event.Event;
import org.apache.flume.event.EventBuilder;
import org.junit.Test;
diff --git a/flume-ng-core/src/test/java/org/apache/flume/interceptor/TestCensoringInterceptor.java b/flume-ng-core/src/test/java/org/apache/flume/interceptor/TestCensoringInterceptor.java
index ce532272..0ce61e6b 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/interceptor/TestCensoringInterceptor.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/interceptor/TestCensoringInterceptor.java
@@ -23,11 +23,11 @@ import java.util.Map;
import org.apache.flume.Channel;
import org.apache.flume.ChannelSelector;
import org.apache.flume.Context;
-import org.apache.flume.Event;
import org.apache.flume.Transaction;
import org.apache.flume.channel.ChannelProcessor;
import org.apache.flume.channel.MemoryChannel;
import org.apache.flume.channel.ReplicatingChannelSelector;
+import org.apache.flume.event.Event;
import org.apache.flume.event.EventBuilder;
import org.junit.Assert;
import org.junit.Test;
diff --git a/flume-ng-core/src/test/java/org/apache/flume/interceptor/TestHostInterceptor.java b/flume-ng-core/src/test/java/org/apache/flume/interceptor/TestHostInterceptor.java
index 92bd7aeb..42dee620 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/interceptor/TestHostInterceptor.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/interceptor/TestHostInterceptor.java
@@ -19,7 +19,7 @@ package org.apache.flume.interceptor;
import com.google.common.base.Charsets;
import java.net.InetAddress;
import org.apache.flume.Context;
-import org.apache.flume.Event;
+import org.apache.flume.event.Event;
import org.apache.flume.event.EventBuilder;
import org.apache.flume.interceptor.HostInterceptor.Constants;
import org.junit.Assert;
diff --git a/flume-ng-core/src/test/java/org/apache/flume/interceptor/TestRegexExtractorInterceptor.java b/flume-ng-core/src/test/java/org/apache/flume/interceptor/TestRegexExtractorInterceptor.java
index cd7f4efc..9dc230b6 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/interceptor/TestRegexExtractorInterceptor.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/interceptor/TestRegexExtractorInterceptor.java
@@ -23,7 +23,7 @@ import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import org.apache.flume.Context;
-import org.apache.flume.Event;
+import org.apache.flume.event.Event;
import org.apache.flume.event.EventBuilder;
import org.apache.flume.interceptor.Interceptor.Builder;
import org.junit.Assert;
diff --git a/flume-ng-core/src/test/java/org/apache/flume/interceptor/TestRegexFilteringInterceptor.java b/flume-ng-core/src/test/java/org/apache/flume/interceptor/TestRegexFilteringInterceptor.java
index f1a1d07b..5920dc29 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/interceptor/TestRegexFilteringInterceptor.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/interceptor/TestRegexFilteringInterceptor.java
@@ -18,7 +18,7 @@ package org.apache.flume.interceptor;
import com.google.common.base.Charsets;
import org.apache.flume.Context;
-import org.apache.flume.Event;
+import org.apache.flume.event.Event;
import org.apache.flume.event.EventBuilder;
import org.apache.flume.interceptor.RegexFilteringInterceptor.Constants;
import org.junit.Assert;
diff --git a/flume-ng-core/src/test/java/org/apache/flume/interceptor/TestSearchAndReplaceInterceptor.java b/flume-ng-core/src/test/java/org/apache/flume/interceptor/TestSearchAndReplaceInterceptor.java
index 3e435897..7b4447d8 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/interceptor/TestSearchAndReplaceInterceptor.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/interceptor/TestSearchAndReplaceInterceptor.java
@@ -20,7 +20,7 @@ import static org.junit.Assert.assertEquals;
import com.google.common.base.Charsets;
import org.apache.flume.Context;
-import org.apache.flume.Event;
+import org.apache.flume.event.Event;
import org.apache.flume.event.EventBuilder;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
diff --git a/flume-ng-core/src/test/java/org/apache/flume/interceptor/TestStaticInterceptor.java b/flume-ng-core/src/test/java/org/apache/flume/interceptor/TestStaticInterceptor.java
index 74698abe..2e981ba0 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/interceptor/TestStaticInterceptor.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/interceptor/TestStaticInterceptor.java
@@ -18,7 +18,7 @@ package org.apache.flume.interceptor;
import com.google.common.base.Charsets;
import org.apache.flume.Context;
-import org.apache.flume.Event;
+import org.apache.flume.event.Event;
import org.apache.flume.event.EventBuilder;
import org.apache.flume.interceptor.StaticInterceptor.Constants;
import org.junit.Assert;
diff --git a/flume-ng-core/src/test/java/org/apache/flume/interceptor/TestTimestampInterceptor.java b/flume-ng-core/src/test/java/org/apache/flume/interceptor/TestTimestampInterceptor.java
index c3ba6627..1196b073 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/interceptor/TestTimestampInterceptor.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/interceptor/TestTimestampInterceptor.java
@@ -18,7 +18,7 @@ package org.apache.flume.interceptor;
import com.google.common.base.Charsets;
import org.apache.flume.Context;
-import org.apache.flume.Event;
+import org.apache.flume.event.Event;
import org.apache.flume.event.EventBuilder;
import org.apache.flume.interceptor.TimestampInterceptor.Constants;
import org.junit.Assert;
diff --git a/flume-ng-core/src/test/java/org/apache/flume/serialization/TestLineDeserializer.java b/flume-ng-core/src/test/java/org/apache/flume/serialization/TestLineDeserializer.java
index 12f09b20..d30323e7 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/serialization/TestLineDeserializer.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/serialization/TestLineDeserializer.java
@@ -21,7 +21,7 @@ import java.io.IOException;
import java.util.List;
import junit.framework.Assert;
import org.apache.flume.Context;
-import org.apache.flume.Event;
+import org.apache.flume.event.Event;
import org.junit.Before;
import org.junit.Test;
diff --git a/flume-ng-core/src/test/java/org/apache/flume/sink/TestFailoverSinkProcessor.java b/flume-ng-core/src/test/java/org/apache/flume/sink/TestFailoverSinkProcessor.java
index 0ce85831..6df9e1b0 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/sink/TestFailoverSinkProcessor.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/sink/TestFailoverSinkProcessor.java
@@ -23,13 +23,13 @@ import java.util.Map;
import junit.framework.Assert;
import org.apache.flume.Channel;
import org.apache.flume.Context;
-import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.Sink;
import org.apache.flume.SinkRunner;
import org.apache.flume.Transaction;
import org.apache.flume.channel.MemoryChannel;
import org.apache.flume.conf.Configurables;
+import org.apache.flume.event.Event;
import org.apache.flume.event.EventBuilder;
import org.apache.flume.lifecycle.LifecycleState;
import org.junit.Test;
diff --git a/flume-ng-core/src/test/java/org/apache/flume/sink/TestLoadBalancingSinkProcessor.java b/flume-ng-core/src/test/java/org/apache/flume/sink/TestLoadBalancingSinkProcessor.java
index 47e86755..5550ad4d 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/sink/TestLoadBalancingSinkProcessor.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/sink/TestLoadBalancingSinkProcessor.java
@@ -26,12 +26,12 @@ import java.util.Set;
import junit.framework.Assert;
import org.apache.flume.Channel;
import org.apache.flume.Context;
-import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.Sink;
import org.apache.flume.Sink.Status;
import org.apache.flume.Transaction;
import org.apache.flume.channel.AbstractChannel;
+import org.apache.flume.event.Event;
import org.apache.flume.exception.ChannelException;
import org.junit.Test;
diff --git a/flume-ng-core/src/test/java/org/apache/flume/sink/TestLoggerSink.java b/flume-ng-core/src/test/java/org/apache/flume/sink/TestLoggerSink.java
index e8ed3a80..ebdb74ea 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/sink/TestLoggerSink.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/sink/TestLoggerSink.java
@@ -19,10 +19,10 @@ package org.apache.flume.sink;
import com.google.common.base.Strings;
import org.apache.flume.Channel;
import org.apache.flume.Context;
-import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.channel.PseudoTxnMemoryChannel;
import org.apache.flume.conf.Configurables;
+import org.apache.flume.event.Event;
import org.apache.flume.event.EventBuilder;
import org.apache.flume.lifecycle.LifecycleException;
import org.junit.Before;
diff --git a/flume-ng-core/src/test/java/org/apache/flume/sink/TestRollingFileSink.java b/flume-ng-core/src/test/java/org/apache/flume/sink/TestRollingFileSink.java
index 2d948b53..28dd151c 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/sink/TestRollingFileSink.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/sink/TestRollingFileSink.java
@@ -22,13 +22,13 @@ import java.io.FileReader;
import java.io.IOException;
import org.apache.flume.Channel;
import org.apache.flume.Context;
-import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.Transaction;
import org.apache.flume.channel.BasicTransactionSemantics;
import org.apache.flume.channel.MemoryChannel;
import org.apache.flume.channel.PseudoTxnMemoryChannel;
import org.apache.flume.conf.Configurables;
+import org.apache.flume.event.Event;
import org.apache.flume.event.SimpleEvent;
import org.apache.flume.exception.ChannelException;
import org.apache.flume.instrumentation.SinkCounter;
diff --git a/flume-ng-core/src/test/java/org/apache/flume/source/TestExecSource.java b/flume-ng-core/src/test/java/org/apache/flume/source/TestExecSource.java
index e007ff73..2213e118 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/source/TestExecSource.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/source/TestExecSource.java
@@ -44,13 +44,13 @@ import org.apache.commons.lang3.SystemUtils;
import org.apache.flume.Channel;
import org.apache.flume.ChannelSelector;
import org.apache.flume.Context;
-import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.Transaction;
import org.apache.flume.channel.ChannelProcessor;
import org.apache.flume.channel.MemoryChannel;
import org.apache.flume.channel.ReplicatingChannelSelector;
import org.apache.flume.conf.Configurables;
+import org.apache.flume.event.Event;
import org.apache.flume.lifecycle.LifecycleException;
import org.junit.After;
import org.junit.Assert;
diff --git a/flume-ng-core/src/test/java/org/apache/flume/source/TestNetcatSource.java b/flume-ng-core/src/test/java/org/apache/flume/source/TestNetcatSource.java
index 0fde49a4..490b1333 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/source/TestNetcatSource.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/source/TestNetcatSource.java
@@ -32,13 +32,13 @@ import org.apache.commons.io.LineIterator;
import org.apache.flume.Channel;
import org.apache.flume.ChannelSelector;
import org.apache.flume.Context;
-import org.apache.flume.Event;
import org.apache.flume.FlumeException;
import org.apache.flume.Transaction;
import org.apache.flume.channel.ChannelProcessor;
import org.apache.flume.channel.MemoryChannel;
import org.apache.flume.channel.ReplicatingChannelSelector;
import org.apache.flume.conf.Configurables;
+import org.apache.flume.event.Event;
import org.apache.flume.lifecycle.LifecycleController;
import org.apache.flume.lifecycle.LifecycleState;
import org.apache.logging.log4j.LogManager;
diff --git a/flume-ng-core/src/test/java/org/apache/flume/source/TestPollableSourceRunner.java b/flume-ng-core/src/test/java/org/apache/flume/source/TestPollableSourceRunner.java
index 6a8200d5..8cc86f7a 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/source/TestPollableSourceRunner.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/source/TestPollableSourceRunner.java
@@ -21,7 +21,6 @@ import java.util.concurrent.CountDownLatch;
import org.apache.flume.Channel;
import org.apache.flume.ChannelSelector;
import org.apache.flume.Context;
-import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.PollableSource;
import org.apache.flume.Transaction;
@@ -29,6 +28,7 @@ import org.apache.flume.channel.ChannelProcessor;
import org.apache.flume.channel.MemoryChannel;
import org.apache.flume.channel.ReplicatingChannelSelector;
import org.apache.flume.conf.Configurables;
+import org.apache.flume.event.Event;
import org.apache.flume.event.EventBuilder;
import org.apache.flume.lifecycle.LifecycleState;
import org.apache.logging.log4j.LogManager;
diff --git a/flume-ng-core/src/test/java/org/apache/flume/source/TestSequenceGeneratorSource.java b/flume-ng-core/src/test/java/org/apache/flume/source/TestSequenceGeneratorSource.java
index c1ce170d..ef4f1e67 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/source/TestSequenceGeneratorSource.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/source/TestSequenceGeneratorSource.java
@@ -22,11 +22,11 @@ import java.util.LinkedHashSet;
import java.util.List;
import java.util.Set;
import org.apache.flume.Context;
-import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.PollableSource;
import org.apache.flume.channel.ChannelProcessor;
import org.apache.flume.conf.Configurables;
+import org.apache.flume.event.Event;
import org.apache.flume.exception.ChannelException;
import org.junit.Assert;
import org.junit.Before;
diff --git a/flume-ng-core/src/test/java/org/apache/flume/source/TestStressSource.java b/flume-ng-core/src/test/java/org/apache/flume/source/TestStressSource.java
index 2f5d159a..65a6e04d 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/source/TestStressSource.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/source/TestStressSource.java
@@ -28,10 +28,10 @@ import junit.framework.Assert;
import junit.framework.TestCase;
import org.apache.flume.Context;
import org.apache.flume.CounterGroup;
-import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.PollableSource.Status;
import org.apache.flume.channel.ChannelProcessor;
+import org.apache.flume.event.Event;
import org.apache.flume.exception.ChannelException;
import org.junit.Before;
import org.junit.Test;
diff --git a/flume-ng-node/src/test/java/org/apache/flume/node/TestAbstractConfigurationProvider.java b/flume-ng-node/src/test/java/org/apache/flume/node/TestAbstractConfigurationProvider.java
index bceb7099..26b26e9b 100644
--- a/flume-ng-node/src/test/java/org/apache/flume/node/TestAbstractConfigurationProvider.java
+++ b/flume-ng-node/src/test/java/org/apache/flume/node/TestAbstractConfigurationProvider.java
@@ -21,7 +21,6 @@ import java.util.Map;
import junit.framework.Assert;
import org.apache.flume.Channel;
import org.apache.flume.Context;
-import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.Transaction;
import org.apache.flume.annotations.Disposable;
@@ -29,6 +28,7 @@ import org.apache.flume.annotations.Recyclable;
import org.apache.flume.channel.AbstractChannel;
import org.apache.flume.conf.Configurable;
import org.apache.flume.conf.FlumeConfiguration;
+import org.apache.flume.event.Event;
import org.apache.flume.exception.ChannelException;
import org.apache.flume.sink.AbstractSink;
import org.apache.flume.source.AbstractSource;
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 f0bce1f5..4c95e63d 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
@@ -34,10 +34,10 @@ 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.Event;
import org.apache.flume.event.SimpleEvent;
import org.apache.flume.lifecycle.LifecycleAware;
import org.apache.flume.lifecycle.LifecycleState;
diff --git a/flume-ng-node/src/test/java/org/apache/flume/sink/NullInitSink.java b/flume-ng-node/src/test/java/org/apache/flume/sink/NullInitSink.java
index 136f36cf..4cb81aec 100644
--- a/flume-ng-node/src/test/java/org/apache/flume/sink/NullInitSink.java
+++ b/flume-ng-node/src/test/java/org/apache/flume/sink/NullInitSink.java
@@ -20,12 +20,12 @@ 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.event.Event;
import org.apache.flume.node.Initializable;
import org.apache.flume.node.MaterializedConfiguration;
import org.apache.flume.source.EventProcessor;
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
index dd2a2dfe..f80b0780 100644
--- 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
@@ -16,7 +16,7 @@
*/
package org.apache.flume.source;
-import org.apache.flume.Event;
+import org.apache.flume.event.Event;
/**
* Interface indicating processEvent is implemented.
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
index 4b359b15..b71ae264 100644
--- 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
@@ -17,9 +17,9 @@
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.event.Event;
import org.apache.flume.instrumentation.SourceCounter;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
diff --git a/flume-ng-node/src/test/java/org/apache/flume/source/TestNetcatSource.java b/flume-ng-node/src/test/java/org/apache/flume/source/TestNetcatSource.java
index b1ba98c2..9d3fe3c7 100644
--- a/flume-ng-node/src/test/java/org/apache/flume/source/TestNetcatSource.java
+++ b/flume-ng-node/src/test/java/org/apache/flume/source/TestNetcatSource.java
@@ -33,7 +33,6 @@ import java.util.concurrent.TimeUnit;
import org.apache.flume.Channel;
import org.apache.flume.ChannelSelector;
import org.apache.flume.Context;
-import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.EventDrivenSource;
import org.apache.flume.Transaction;
@@ -41,6 +40,7 @@ import org.apache.flume.channel.ChannelProcessor;
import org.apache.flume.channel.MemoryChannel;
import org.apache.flume.channel.ReplicatingChannelSelector;
import org.apache.flume.conf.Configurables;
+import org.apache.flume.event.Event;
import org.apache.flume.lifecycle.LifecycleException;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
diff --git a/flume-ng-sdk/pom.xml b/flume-ng-sdk/pom.xml
index b46565d8..76286ef9 100644
--- a/flume-ng-sdk/pom.xml
+++ b/flume-ng-sdk/pom.xml
@@ -34,13 +34,18 @@
<flume.baseline.groupId>org.apache.flume</flume.baseline.groupId>
<flume.baseline.artifactId>flume-ng-sdk</flume.baseline.artifactId>
<!-- TODO fix spotbugs/pmd violations -->
- <spotbugs.maxAllowedViolations>69</spotbugs.maxAllowedViolations>
- <pmd.maxAllowedViolations>30</pmd.maxAllowedViolations>
+ <spotbugs.maxAllowedViolations>5</spotbugs.maxAllowedViolations>
+ <pmd.maxAllowedViolations>20</pmd.maxAllowedViolations>
<module.name>org.apache.flume.sdk</module.name>
</properties>
<dependencies>
+ <dependency>
+ <groupId>org.apache.flume</groupId>
+ <artifactId>flume-client</artifactId>
+ </dependency>
+
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
diff --git a/flume-ng-sdk/src/main/java/org/apache/flume/serialization/EventDeserializer.java b/flume-ng-sdk/src/main/java/org/apache/flume/serialization/EventDeserializer.java
index f616f2fb..5e3c354b 100644
--- a/flume-ng-sdk/src/main/java/org/apache/flume/serialization/EventDeserializer.java
+++ b/flume-ng-sdk/src/main/java/org/apache/flume/serialization/EventDeserializer.java
@@ -20,9 +20,9 @@ import java.io.Closeable;
import java.io.IOException;
import java.util.List;
import org.apache.flume.Context;
-import org.apache.flume.Event;
import org.apache.flume.annotations.InterfaceAudience;
import org.apache.flume.annotations.InterfaceStability;
+import org.apache.flume.event.Event;
/**
* Establishes a contract for reading events stored in arbitrary formats from
diff --git a/flume-ng-sdk/src/main/java/org/apache/flume/event/EventReader.java b/flume-ng-sdk/src/main/java/org/apache/flume/serialization/EventReader.java
similarity index 96%
rename from flume-ng-sdk/src/main/java/org/apache/flume/event/EventReader.java
rename to flume-ng-sdk/src/main/java/org/apache/flume/serialization/EventReader.java
index b6acc287..c275294c 100644
--- a/flume-ng-sdk/src/main/java/org/apache/flume/event/EventReader.java
+++ b/flume-ng-sdk/src/main/java/org/apache/flume/serialization/EventReader.java
@@ -14,14 +14,14 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.flume.event;
+package org.apache.flume.serialization;
import java.io.Closeable;
import java.io.IOException;
import java.util.List;
-import org.apache.flume.Event;
import org.apache.flume.annotations.InterfaceAudience;
import org.apache.flume.annotations.InterfaceStability;
+import org.apache.flume.event.Event;
/**
* A line reader produces a stream of lines for the {@link AvroCLIClient} to
diff --git a/flume-ng-sdk/src/main/java/org/apache/flume/serialization/LineDeserializer.java b/flume-ng-sdk/src/main/java/org/apache/flume/serialization/LineDeserializer.java
index ce52d611..2745b7f7 100644
--- a/flume-ng-sdk/src/main/java/org/apache/flume/serialization/LineDeserializer.java
+++ b/flume-ng-sdk/src/main/java/org/apache/flume/serialization/LineDeserializer.java
@@ -21,9 +21,9 @@ import java.io.IOException;
import java.nio.charset.Charset;
import java.util.List;
import org.apache.flume.Context;
-import org.apache.flume.Event;
import org.apache.flume.annotations.InterfaceAudience;
import org.apache.flume.annotations.InterfaceStability;
+import org.apache.flume.event.Event;
import org.apache.flume.event.EventBuilder;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
diff --git a/flume-ng-sdk/src/main/java/org/apache/flume/event/ReliableEventReader.java b/flume-ng-sdk/src/main/java/org/apache/flume/serialization/ReliableEventReader.java
similarity index 97%
rename from flume-ng-sdk/src/main/java/org/apache/flume/event/ReliableEventReader.java
rename to flume-ng-sdk/src/main/java/org/apache/flume/serialization/ReliableEventReader.java
index 3102da30..aeea9ddb 100644
--- a/flume-ng-sdk/src/main/java/org/apache/flume/event/ReliableEventReader.java
+++ b/flume-ng-sdk/src/main/java/org/apache/flume/serialization/ReliableEventReader.java
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.flume.event;
+package org.apache.flume.serialization;
import java.io.IOException;
import org.apache.flume.annotations.InterfaceAudience;
diff --git a/flume-ng-sdk/src/main/java/org/apache/flume/event/SimpleTextLineEventReader.java b/flume-ng-sdk/src/main/java/org/apache/flume/serialization/SimpleTextLineEventReader.java
similarity index 94%
rename from flume-ng-sdk/src/main/java/org/apache/flume/event/SimpleTextLineEventReader.java
rename to flume-ng-sdk/src/main/java/org/apache/flume/serialization/SimpleTextLineEventReader.java
index 10fa1205..a256939c 100644
--- a/flume-ng-sdk/src/main/java/org/apache/flume/event/SimpleTextLineEventReader.java
+++ b/flume-ng-sdk/src/main/java/org/apache/flume/serialization/SimpleTextLineEventReader.java
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.flume.event;
+package org.apache.flume.serialization;
import com.google.common.base.Charsets;
import com.google.common.collect.Lists;
@@ -22,9 +22,10 @@ import java.io.BufferedReader;
import java.io.IOException;
import java.io.Reader;
import java.util.List;
-import org.apache.flume.Event;
import org.apache.flume.annotations.InterfaceAudience;
import org.apache.flume.annotations.InterfaceStability;
+import org.apache.flume.event.Event;
+import org.apache.flume.event.EventBuilder;
/**
* A {@link EventReader} implementation which delegates to a
diff --git a/flume-ng-sdk/src/test/java/org/apache/flume/event/TestBufferedLineReader.java b/flume-ng-sdk/src/test/java/org/apache/flume/serialization/TestBufferedLineReader.java
similarity index 98%
rename from flume-ng-sdk/src/test/java/org/apache/flume/event/TestBufferedLineReader.java
rename to flume-ng-sdk/src/test/java/org/apache/flume/serialization/TestBufferedLineReader.java
index ac60388a..9ab0c23f 100644
--- a/flume-ng-sdk/src/test/java/org/apache/flume/event/TestBufferedLineReader.java
+++ b/flume-ng-sdk/src/test/java/org/apache/flume/serialization/TestBufferedLineReader.java
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.flume.event;
+package org.apache.flume.serialization;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
@@ -26,7 +26,7 @@ import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.util.List;
-import org.apache.flume.Event;
+import org.apache.flume.event.Event;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
diff --git a/flume-ng-sources/flume-netcat-source/src/main/java/org/apache/flume/source/netcat/NetcatUdpSource.java b/flume-ng-sources/flume-netcat-source/src/main/java/org/apache/flume/source/netcat/NetcatUdpSource.java
index 23b3d6b5..b4a25679 100644
--- a/flume-ng-sources/flume-netcat-source/src/main/java/org/apache/flume/source/netcat/NetcatUdpSource.java
+++ b/flume-ng-sources/flume-netcat-source/src/main/java/org/apache/flume/source/netcat/NetcatUdpSource.java
@@ -34,10 +34,10 @@ import java.util.HashMap;
import java.util.Map;
import org.apache.flume.Context;
import org.apache.flume.CounterGroup;
-import org.apache.flume.Event;
import org.apache.flume.EventDrivenSource;
import org.apache.flume.conf.Configurable;
import org.apache.flume.conf.Configurables;
+import org.apache.flume.event.Event;
import org.apache.flume.event.EventBuilder;
import org.apache.flume.exception.ChannelException;
import org.apache.flume.source.AbstractSource;
diff --git a/flume-ng-sources/flume-netcat-source/src/test/java/org/apache/flume/source/netcat/TestNetcatUdpSource.java b/flume-ng-sources/flume-netcat-source/src/test/java/org/apache/flume/source/netcat/TestNetcatUdpSource.java
index 191c0e9b..013c0536 100644
--- a/flume-ng-sources/flume-netcat-source/src/test/java/org/apache/flume/source/netcat/TestNetcatUdpSource.java
+++ b/flume-ng-sources/flume-netcat-source/src/test/java/org/apache/flume/source/netcat/TestNetcatUdpSource.java
@@ -26,12 +26,12 @@ import java.util.List;
import org.apache.flume.Channel;
import org.apache.flume.ChannelSelector;
import org.apache.flume.Context;
-import org.apache.flume.Event;
import org.apache.flume.Transaction;
import org.apache.flume.channel.ChannelProcessor;
import org.apache.flume.channel.MemoryChannel;
import org.apache.flume.channel.ReplicatingChannelSelector;
import org.apache.flume.conf.Configurables;
+import org.apache.flume.event.Event;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.junit.Assert;
diff --git a/flume-ng-sources/flume-syslog-source/src/main/java/org/apache/flume/source/syslog/MultiportSyslogTCPSource.java b/flume-ng-sources/flume-syslog-source/src/main/java/org/apache/flume/source/syslog/MultiportSyslogTCPSource.java
index 6cf35285..411e50c8 100644
--- a/flume-ng-sources/flume-syslog-source/src/main/java/org/apache/flume/source/syslog/MultiportSyslogTCPSource.java
+++ b/flume-ng-sources/flume-syslog-source/src/main/java/org/apache/flume/source/syslog/MultiportSyslogTCPSource.java
@@ -30,12 +30,12 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import javax.net.ssl.SSLParameters;
import org.apache.flume.Context;
-import org.apache.flume.Event;
import org.apache.flume.EventDrivenSource;
import org.apache.flume.channel.ChannelProcessor;
import org.apache.flume.conf.BatchSizeSupported;
import org.apache.flume.conf.Configurable;
import org.apache.flume.conf.LogPrivacyUtil;
+import org.apache.flume.event.Event;
import org.apache.flume.event.EventBuilder;
import org.apache.flume.instrumentation.SourceCounter;
import org.apache.flume.source.SslContextAwareAbstractSource;
diff --git a/flume-ng-sources/flume-syslog-source/src/main/java/org/apache/flume/source/syslog/SyslogParser.java b/flume-ng-sources/flume-syslog-source/src/main/java/org/apache/flume/source/syslog/SyslogParser.java
index d41da4f1..6664d6fb 100644
--- a/flume-ng-sources/flume-syslog-source/src/main/java/org/apache/flume/source/syslog/SyslogParser.java
+++ b/flume-ng-sources/flume-syslog-source/src/main/java/org/apache/flume/source/syslog/SyslogParser.java
@@ -37,9 +37,9 @@ import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.regex.Pattern;
import org.apache.commons.lang3.StringUtils;
-import org.apache.flume.Event;
import org.apache.flume.annotations.InterfaceAudience;
import org.apache.flume.annotations.InterfaceStability;
+import org.apache.flume.event.Event;
import org.apache.flume.event.EventBuilder;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
diff --git a/flume-ng-sources/flume-syslog-source/src/main/java/org/apache/flume/source/syslog/SyslogTcpSource.java b/flume-ng-sources/flume-syslog-source/src/main/java/org/apache/flume/source/syslog/SyslogTcpSource.java
index a7186ca1..5c038730 100644
--- a/flume-ng-sources/flume-syslog-source/src/main/java/org/apache/flume/source/syslog/SyslogTcpSource.java
+++ b/flume-ng-sources/flume-syslog-source/src/main/java/org/apache/flume/source/syslog/SyslogTcpSource.java
@@ -39,10 +39,10 @@ import java.util.Optional;
import java.util.Set;
import javax.net.ssl.SSLEngine;
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.conf.Configurables;
+import org.apache.flume.event.Event;
import org.apache.flume.exception.ChannelException;
import org.apache.flume.instrumentation.SourceCounter;
import org.apache.flume.source.SslContextAwareAbstractSource;
diff --git a/flume-ng-sources/flume-syslog-source/src/main/java/org/apache/flume/source/syslog/SyslogUDPSource.java b/flume-ng-sources/flume-syslog-source/src/main/java/org/apache/flume/source/syslog/SyslogUDPSource.java
index bec59177..94f03fc5 100644
--- a/flume-ng-sources/flume-syslog-source/src/main/java/org/apache/flume/source/syslog/SyslogUDPSource.java
+++ b/flume-ng-sources/flume-syslog-source/src/main/java/org/apache/flume/source/syslog/SyslogUDPSource.java
@@ -31,10 +31,10 @@ import java.net.SocketAddress;
import java.util.Map;
import java.util.Set;
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.conf.Configurables;
+import org.apache.flume.event.Event;
import org.apache.flume.exception.ChannelException;
import org.apache.flume.instrumentation.SourceCounter;
import org.apache.flume.source.AbstractSource;
diff --git a/flume-ng-sources/flume-syslog-source/src/main/java/org/apache/flume/source/syslog/SyslogUtils.java b/flume-ng-sources/flume-syslog-source/src/main/java/org/apache/flume/source/syslog/SyslogUtils.java
index 3ac1943d..d2a755a0 100644
--- a/flume-ng-sources/flume-syslog-source/src/main/java/org/apache/flume/source/syslog/SyslogUtils.java
+++ b/flume-ng-sources/flume-syslog-source/src/main/java/org/apache/flume/source/syslog/SyslogUtils.java
@@ -37,9 +37,9 @@ import java.util.Set;
import java.util.regex.MatchResult;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
-import org.apache.flume.Event;
import org.apache.flume.annotations.InterfaceAudience;
import org.apache.flume.annotations.InterfaceStability;
+import org.apache.flume.event.Event;
import org.apache.flume.event.EventBuilder;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
diff --git a/flume-ng-sources/flume-syslog-source/src/test/java/org/apache/flume/source/syslog/TestMultiportSyslogTCPSource.java b/flume-ng-sources/flume-syslog-source/src/test/java/org/apache/flume/source/syslog/TestMultiportSyslogTCPSource.java
index af739b58..a1b62ee6 100644
--- a/flume-ng-sources/flume-syslog-source/src/test/java/org/apache/flume/source/syslog/TestMultiportSyslogTCPSource.java
+++ b/flume-ng-sources/flume-syslog-source/src/test/java/org/apache/flume/source/syslog/TestMultiportSyslogTCPSource.java
@@ -54,12 +54,12 @@ import javax.net.ssl.X509TrustManager;
import org.apache.flume.Channel;
import org.apache.flume.ChannelSelector;
import org.apache.flume.Context;
-import org.apache.flume.Event;
import org.apache.flume.Transaction;
import org.apache.flume.channel.ChannelProcessor;
import org.apache.flume.channel.MemoryChannel;
import org.apache.flume.channel.ReplicatingChannelSelector;
import org.apache.flume.conf.Configurables;
+import org.apache.flume.event.Event;
import org.apache.flume.exception.ChannelException;
import org.apache.flume.instrumentation.SourceCounter;
import org.apache.flume.sdk.test.TestKeyStores;
diff --git a/flume-ng-sources/flume-syslog-source/src/test/java/org/apache/flume/source/syslog/TestSyslogParser.java b/flume-ng-sources/flume-syslog-source/src/test/java/org/apache/flume/source/syslog/TestSyslogParser.java
index ca6e53ed..5a291cec 100644
--- a/flume-ng-sources/flume-syslog-source/src/test/java/org/apache/flume/source/syslog/TestSyslogParser.java
+++ b/flume-ng-sources/flume-syslog-source/src/test/java/org/apache/flume/source/syslog/TestSyslogParser.java
@@ -26,7 +26,7 @@ import java.time.ZoneOffset;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
-import org.apache.flume.Event;
+import org.apache.flume.event.Event;
import org.junit.Assert;
import org.junit.Test;
diff --git a/flume-ng-sources/flume-syslog-source/src/test/java/org/apache/flume/source/syslog/TestSyslogTcpSource.java b/flume-ng-sources/flume-syslog-source/src/test/java/org/apache/flume/source/syslog/TestSyslogTcpSource.java
index a4eb816f..3b9daefc 100644
--- a/flume-ng-sources/flume-syslog-source/src/test/java/org/apache/flume/source/syslog/TestSyslogTcpSource.java
+++ b/flume-ng-sources/flume-syslog-source/src/test/java/org/apache/flume/source/syslog/TestSyslogTcpSource.java
@@ -43,12 +43,12 @@ import javax.net.ssl.X509TrustManager;
import org.apache.flume.Channel;
import org.apache.flume.ChannelSelector;
import org.apache.flume.Context;
-import org.apache.flume.Event;
import org.apache.flume.Transaction;
import org.apache.flume.channel.ChannelProcessor;
import org.apache.flume.channel.MemoryChannel;
import org.apache.flume.channel.ReplicatingChannelSelector;
import org.apache.flume.conf.Configurables;
+import org.apache.flume.event.Event;
import org.apache.flume.exception.ChannelException;
import org.apache.flume.sdk.test.TestKeyStores;
import org.apache.logging.log4j.LogManager;
diff --git a/flume-ng-sources/flume-syslog-source/src/test/java/org/apache/flume/source/syslog/TestSyslogUdpSource.java b/flume-ng-sources/flume-syslog-source/src/test/java/org/apache/flume/source/syslog/TestSyslogUdpSource.java
index e9ade851..86cea971 100644
--- a/flume-ng-sources/flume-syslog-source/src/test/java/org/apache/flume/source/syslog/TestSyslogUdpSource.java
+++ b/flume-ng-sources/flume-syslog-source/src/test/java/org/apache/flume/source/syslog/TestSyslogUdpSource.java
@@ -35,12 +35,12 @@ import java.util.Map;
import org.apache.flume.Channel;
import org.apache.flume.ChannelSelector;
import org.apache.flume.Context;
-import org.apache.flume.Event;
import org.apache.flume.Transaction;
import org.apache.flume.channel.ChannelProcessor;
import org.apache.flume.channel.MemoryChannel;
import org.apache.flume.channel.ReplicatingChannelSelector;
import org.apache.flume.conf.Configurables;
+import org.apache.flume.event.Event;
import org.apache.flume.exception.ChannelException;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
diff --git a/flume-ng-sources/flume-syslog-source/src/test/java/org/apache/flume/source/syslog/TestSyslogUtils.java b/flume-ng-sources/flume-syslog-source/src/test/java/org/apache/flume/source/syslog/TestSyslogUtils.java
index df2ef51d..eb9c17f6 100644
--- a/flume-ng-sources/flume-syslog-source/src/test/java/org/apache/flume/source/syslog/TestSyslogUtils.java
+++ b/flume-ng-sources/flume-syslog-source/src/test/java/org/apache/flume/source/syslog/TestSyslogUtils.java
@@ -33,7 +33,7 @@ import java.util.Date;
import java.util.HashSet;
import java.util.Locale;
import java.util.Map;
-import org.apache.flume.Event;
+import org.apache.flume.event.Event;
import org.junit.Assert;
import org.junit.Test;
diff --git a/flume-ng-sources/flume-taildir-source/src/main/java/org/apache/flume/source/taildir/ReliableTaildirEventReader.java b/flume-ng-sources/flume-taildir-source/src/main/java/org/apache/flume/source/taildir/ReliableTaildirEventReader.java
index e3754d77..094bfab4 100644
--- a/flume-ng-sources/flume-taildir-source/src/main/java/org/apache/flume/source/taildir/ReliableTaildirEventReader.java
+++ b/flume-ng-sources/flume-taildir-source/src/main/java/org/apache/flume/source/taildir/ReliableTaildirEventReader.java
@@ -32,11 +32,11 @@ import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
-import org.apache.flume.Event;
import org.apache.flume.FlumeException;
import org.apache.flume.annotations.InterfaceAudience;
import org.apache.flume.annotations.InterfaceStability;
-import org.apache.flume.event.ReliableEventReader;
+import org.apache.flume.event.Event;
+import org.apache.flume.serialization.ReliableEventReader;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
diff --git a/flume-ng-sources/flume-taildir-source/src/main/java/org/apache/flume/source/taildir/TailFile.java b/flume-ng-sources/flume-taildir-source/src/main/java/org/apache/flume/source/taildir/TailFile.java
index 7823a952..cc7063ad 100644
--- a/flume-ng-sources/flume-taildir-source/src/main/java/org/apache/flume/source/taildir/TailFile.java
+++ b/flume-ng-sources/flume-taildir-source/src/main/java/org/apache/flume/source/taildir/TailFile.java
@@ -24,7 +24,7 @@ import java.io.IOException;
import java.io.RandomAccessFile;
import java.util.List;
import java.util.Map;
-import org.apache.flume.Event;
+import org.apache.flume.event.Event;
import org.apache.flume.event.EventBuilder;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
diff --git a/flume-ng-sources/flume-taildir-source/src/main/java/org/apache/flume/source/taildir/TaildirSource.java b/flume-ng-sources/flume-taildir-source/src/main/java/org/apache/flume/source/taildir/TaildirSource.java
index 8d9c0ef9..876b5d15 100644
--- a/flume-ng-sources/flume-taildir-source/src/main/java/org/apache/flume/source/taildir/TaildirSource.java
+++ b/flume-ng-sources/flume-taildir-source/src/main/java/org/apache/flume/source/taildir/TaildirSource.java
@@ -64,11 +64,11 @@ import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import org.apache.flume.Context;
-import org.apache.flume.Event;
import org.apache.flume.FlumeException;
import org.apache.flume.PollableSource;
import org.apache.flume.conf.BatchSizeSupported;
import org.apache.flume.conf.Configurable;
+import org.apache.flume.event.Event;
import org.apache.flume.exception.ChannelException;
import org.apache.flume.instrumentation.SourceCounter;
import org.apache.flume.source.AbstractSource;
diff --git a/flume-ng-sources/flume-taildir-source/src/test/java/org/apache/flume/source/taildir/TestTaildirEventReader.java b/flume-ng-sources/flume-taildir-source/src/test/java/org/apache/flume/source/taildir/TestTaildirEventReader.java
index 54d57380..65bebca8 100644
--- a/flume-ng-sources/flume-taildir-source/src/test/java/org/apache/flume/source/taildir/TestTaildirEventReader.java
+++ b/flume-ng-sources/flume-taildir-source/src/test/java/org/apache/flume/source/taildir/TestTaildirEventReader.java
@@ -35,7 +35,7 @@ import java.nio.file.FileSystems;
import java.util.List;
import java.util.Map;
import java.util.Set;
-import org.apache.flume.Event;
+import org.apache.flume.event.Event;
import org.junit.After;
import org.junit.Assume;
import org.junit.Before;
diff --git a/flume-ng-sources/flume-taildir-source/src/test/java/org/apache/flume/source/taildir/TestTaildirSource.java b/flume-ng-sources/flume-taildir-source/src/test/java/org/apache/flume/source/taildir/TestTaildirSource.java
index ff4bb18f..528d1d72 100644
--- a/flume-ng-sources/flume-taildir-source/src/test/java/org/apache/flume/source/taildir/TestTaildirSource.java
+++ b/flume-ng-sources/flume-taildir-source/src/test/java/org/apache/flume/source/taildir/TestTaildirSource.java
@@ -47,13 +47,13 @@ import java.util.concurrent.TimeUnit;
import org.apache.flume.Channel;
import org.apache.flume.ChannelSelector;
import org.apache.flume.Context;
-import org.apache.flume.Event;
import org.apache.flume.PollableSource.Status;
import org.apache.flume.Transaction;
import org.apache.flume.channel.ChannelProcessor;
import org.apache.flume.channel.MemoryChannel;
import org.apache.flume.channel.ReplicatingChannelSelector;
import org.apache.flume.conf.Configurables;
+import org.apache.flume.event.Event;
import org.apache.flume.exception.ChannelException;
import org.apache.flume.lifecycle.LifecycleController;
import org.apache.flume.lifecycle.LifecycleState;
diff --git a/flume-tools/src/main/java/org/apache/flume/tools/EventValidator.java b/flume-tools/src/main/java/org/apache/flume/tools/EventValidator.java
index d991004d..3d28931a 100644
--- a/flume-tools/src/main/java/org/apache/flume/tools/EventValidator.java
+++ b/flume-tools/src/main/java/org/apache/flume/tools/EventValidator.java
@@ -16,8 +16,8 @@
*/
package org.apache.flume.tools;
-import org.apache.flume.Event;
import org.apache.flume.conf.Configurable;
+import org.apache.flume.event.Event;
/**
* Event Validator interface to be used for validating Events
diff --git a/flume-tools/src/main/java/org/apache/flume/tools/FileChannelIntegrityTool.java b/flume-tools/src/main/java/org/apache/flume/tools/FileChannelIntegrityTool.java
index 64c1a8b4..1bbf4efe 100644
--- a/flume-tools/src/main/java/org/apache/flume/tools/FileChannelIntegrityTool.java
+++ b/flume-tools/src/main/java/org/apache/flume/tools/FileChannelIntegrityTool.java
@@ -32,7 +32,6 @@ import org.apache.commons.cli.OptionBuilder;
import org.apache.commons.cli.Options;
import org.apache.commons.cli.ParseException;
import org.apache.flume.Context;
-import org.apache.flume.Event;
import org.apache.flume.FlumeException;
import org.apache.flume.channel.file.CorruptEventException;
import org.apache.flume.channel.file.EventUtils;
@@ -42,6 +41,7 @@ import org.apache.flume.channel.file.LogFileV3;
import org.apache.flume.channel.file.LogRecord;
import org.apache.flume.channel.file.Serialization;
import org.apache.flume.channel.file.TransactionEventRecord;
+import org.apache.flume.event.Event;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
diff --git a/flume-tools/src/test/java/org/apache/flume/tools/TestFileChannelIntegrityTool.java b/flume-tools/src/test/java/org/apache/flume/tools/TestFileChannelIntegrityTool.java
index 91da39c0..52fd3b5f 100644
--- a/flume-tools/src/test/java/org/apache/flume/tools/TestFileChannelIntegrityTool.java
+++ b/flume-tools/src/test/java/org/apache/flume/tools/TestFileChannelIntegrityTool.java
@@ -28,7 +28,6 @@ import java.util.Random;
import java.util.Set;
import org.apache.commons.io.FileUtils;
import org.apache.flume.Context;
-import org.apache.flume.Event;
import org.apache.flume.Transaction;
import org.apache.flume.channel.file.FileChannel;
import org.apache.flume.channel.file.FileChannelConfiguration;
@@ -38,6 +37,7 @@ import org.apache.flume.channel.file.LogFileV3;
import org.apache.flume.channel.file.LogRecord;
import org.apache.flume.channel.file.Serialization;
import org.apache.flume.channel.file.WriteOrderOracle;
+import org.apache.flume.event.Event;
import org.apache.flume.event.EventBuilder;
import org.junit.After;
import org.junit.AfterClass;
diff --git a/pom.xml b/pom.xml
index 3dc5bd0f..c8aed0d1 100644
--- a/pom.xml
+++ b/pom.xml
@@ -69,6 +69,7 @@
<module>flume-ng-dist</module>
-->
<module>flume-ng-channels</module>
+ <module>flume-client</module>
<module>flume-ng-sdk</module>
<module>flume-ng-sdk-test</module>
<!-- Temporarily disable integration tests
diff --git a/src/changelog/.2.x.x/add_flume_client.xml b/src/changelog/.2.x.x/add_flume_client.xml
new file mode 100644
index 00000000..efae12b6
--- /dev/null
+++ b/src/changelog/.2.x.x/add_flume_client.xml
@@ -0,0 +1,7 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<entry xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xmlns="https://logging.apache.org/xml/ns"
+ xsi:schemaLocation="https://logging.apache.org/xml/ns https://logging.apache.org/xml/ns/log4j-changelog-0.xsd"
+ type="changed">
+ <description format="asciidoc">Move the client API to the new `flume-client` artifact, so client applications no longer depend on the plugin API. `Event` moves to `org.apache.flume.event`, `RpcClient` extends `Closeable` and reports failures with `IOException` and the new `RpcDeliveryException` instead of `EventDeliveryException` and `FlumeException`</description>
+</entry>
diff --git a/src/site/antora/modules/ROOT/pages/FlumeDeveloperGuide.adoc b/src/site/antora/modules/ROOT/pages/FlumeDeveloperGuide.adoc
index 3b4cce04..b6fc5021 100644
--- a/src/site/antora/modules/ROOT/pages/FlumeDeveloperGuide.adoc
+++ b/src/site/antora/modules/ROOT/pages/FlumeDeveloperGuide.adoc
@@ -164,15 +164,16 @@ within a user's data-generating application:
[source,java]
----
-import org.apache.flume.Event;
-import org.apache.flume.EventDeliveryException;
+import java.io.IOException;
+import java.nio.charset.Charset;
import org.apache.flume.api.RpcClient;
import org.apache.flume.api.RpcClientFactory;
+import org.apache.flume.api.RpcDeliveryException;
+import org.apache.flume.event.Event;
import org.apache.flume.event.EventBuilder;
-import java.nio.charset.Charset;
public class MyApp {
- public static void main(String[] args) {
+ public static void main(String[] args) throws IOException {
MyRpcClientFacade client = new MyRpcClientFacade();
// Initialize client with the remote Flume agent's host and port
client.init("host.example.org", 41414);
@@ -193,7 +194,7 @@ class MyRpcClientFacade {
private String hostname;
private int port;
- public void init(String hostname, int port) {
+ public void init(String hostname, int port) throws IOException {
// Setup the RPC connection
this.hostname = hostname;
this.port = port;
@@ -202,14 +203,14 @@ class MyRpcClientFacade {
// this.client = RpcClientFactory.getThriftInstance(hostname, port);
}
- public void sendDataToFlume(String data) {
+ public void sendDataToFlume(String data) throws IOException {
// Create a Flume Event object that encapsulates the sample data
Event event = EventBuilder.withBody(data, Charset.forName("UTF-8"));
// Send the event
try {
client.append(event);
- } catch (EventDeliveryException e) {
+ } catch (RpcDeliveryException e) {
// clean up and recreate the client
client.close();
client = null;
@@ -219,7 +220,7 @@ class MyRpcClientFacade {
}
}
- public void cleanUp() {
+ public void cleanUp() throws IOException {
// Close the RPC connection
client.close();
}
@@ -291,17 +292,18 @@ within a user's data-generating application:
[source,java]
----
-import org.apache.flume.Event;
-import org.apache.flume.EventDeliveryException;
-import org.apache.flume.event.EventBuilder;
-import org.apache.flume.api.SecureRpcClientFactory;
-import org.apache.flume.api.RpcClientConfigurationConstants;
-import org.apache.flume.api.RpcClient;
+import java.io.IOException;
import java.nio.charset.Charset;
import java.util.Properties;
+import org.apache.flume.api.RpcClient;
+import org.apache.flume.api.RpcClientConfigurationConstants;
+import org.apache.flume.api.RpcDeliveryException;
+import org.apache.flume.api.SecureRpcClientFactory;
+import org.apache.flume.event.Event;
+import org.apache.flume.event.EventBuilder;
public class MyApp {
- public static void main(String[] args) {
+ public static void main(String[] args) throws IOException {
MySecureRpcClientFacade client = new MySecureRpcClientFacade();
// Initialize client with the remote Flume agent's host, port
Properties props = new Properties();
@@ -331,21 +333,21 @@ class MySecureRpcClientFacade {
private RpcClient client;
private Properties properties;
- public void init(Properties properties) {
+ public void init(Properties properties) throws IOException {
// Setup the RPC connection
this.properties = properties;
// Create the ThriftSecureRpcClient instance by using SecureRpcClientFactory
this.client = SecureRpcClientFactory.getThriftInstance(properties);
}
- public void sendDataToFlume(String data) {
+ public void sendDataToFlume(String data) throws IOException {
// Create a Flume Event object that encapsulates the sample data
Event event = EventBuilder.withBody(data, Charset.forName("UTF-8"));
// Send the event
try {
client.append(event);
- } catch (EventDeliveryException e) {
+ } catch (RpcDeliveryException e) {
// clean up and recreate the client
client.close();
client = null;
@@ -353,7 +355,7 @@ class MySecureRpcClientFacade {
}
}
- public void cleanUp() {
+ public void cleanUp() throws IOException {
// Close the RPC connection
client.close();
}