Update to Spring Boot 4 (#8)
* Update to Spring Boot 4
* Use Flume internal SuppressFBWarnings annotation
Replace the LGPL-licensed edu.umd.cs.findbugs SuppressFBWarnings
annotation with Flume's internal org.apache.flume.conf.internal
SuppressFBWarnings, and drop the now-unused spotbugs-annotations
dependency.
Assisted-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
* Set version to 2.0.0-SNAPSHOT and import third-party deps BOM
Change the project version from 2.1.0-SNAPSHOT to 2.0.0-SNAPSHOT and
import the flume-third-party-dependencies BOM in dependencyManagement.
Assisted-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---------
Co-authored-by: Piotr P. Karwasz <piotr@github.copernik.eu>
diff --git a/README.md b/README.md
index 26a4e14..39fd233 100644
--- a/README.md
+++ b/README.md
@@ -37,7 +37,19 @@
Documentation is included in the binary distribution under the docs directory.
In source form, it can be found in the flume-ng-doc directory.
-The Flume 1.x guide and FAQ are available here:
+In general, creating a Flume Spring Boot application requires creating a simple
+project that contains the creation of the Sources, Sinks, and Channels with
+the Sinks encapsulated in SinkRunners and the Sources in SourceRunners and then
+then providing the bootstrap.yml and application.yml files. The Java class
+containing the configuration must be named in the file:
+META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports
+where each line in the file should contain the fully qualified class name of the
+class to be autoconfigured.Note that Spring recommends that classes listed for
+AutoConfiguration should NOT specify packages for component scanning. Instead,
+the specific classes should be specified with @Import or they could also
+be added the the AutoConfiguration import file.
+
+The Flume 2.x guide and FAQ are available here:
* https://cwiki.apache.org/FLUME
* https://cwiki.apache.org/confluence/display/FLUME/Getting+Started
@@ -55,5 +67,5 @@
Compiling Flume Spring Boot requires the following tools:
-* Oracle Java JDK 11
+* Oracle Java JDK 17
* Apache Maven 3.x
diff --git a/flume-spring-boot-dist/pom.xml b/flume-spring-boot-dist/pom.xml
index b3b3d32..1a12c3b 100644
--- a/flume-spring-boot-dist/pom.xml
+++ b/flume-spring-boot-dist/pom.xml
@@ -23,7 +23,7 @@
<parent>
<groupId>org.apache.flume</groupId>
<artifactId>flume-spring-boot-parent</artifactId>
- <version>2.1.0-SNAPSHOT.0</version>
+ <version>2.0.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
diff --git a/flume-spring-boot/pom.xml b/flume-spring-boot/pom.xml
index b85e5e9..bfdefc5 100644
--- a/flume-spring-boot/pom.xml
+++ b/flume-spring-boot/pom.xml
@@ -20,7 +20,7 @@
<parent>
<groupId>org.apache.flume</groupId>
<artifactId>flume-spring-boot-parent</artifactId>
- <version>2.1.0-SNAPSHOT</version>
+ <version>2.0.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
@@ -76,7 +76,6 @@
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
-
<dependency>
<groupId>com.google.code.gson</groupId>
<artifactId>gson</artifactId>
@@ -112,13 +111,7 @@
<dependency>
<groupId>org.apache.logging.log4j</groupId>
- <artifactId>log4j-1.2-api</artifactId>
- <scope>test</scope>
- </dependency>
-
- <dependency>
- <groupId>org.apache.logging.log4j</groupId>
- <artifactId>log4j-slf4j-impl</artifactId>
+ <artifactId>log4j-slf4j2-impl</artifactId>
<scope>test</scope>
</dependency>
<dependency>
diff --git a/flume-spring-boot/src/main/java/org/apache/flume/spring/boot/config/AbstractFlumeConfiguration.java b/flume-spring-boot/src/main/java/org/apache/flume/spring/boot/config/AbstractFlumeConfiguration.java
index 57a567f..962917b 100644
--- a/flume-spring-boot/src/main/java/org/apache/flume/spring/boot/config/AbstractFlumeConfiguration.java
+++ b/flume-spring-boot/src/main/java/org/apache/flume/spring/boot/config/AbstractFlumeConfiguration.java
@@ -17,7 +17,9 @@
package org.apache.flume.spring.boot.config;
import java.util.Arrays;
+import java.util.HashMap;
import java.util.List;
+import java.util.Locale;
import java.util.Map;
import org.apache.flume.Channel;
import org.apache.flume.ChannelSelector;
@@ -29,7 +31,10 @@
import org.apache.flume.Source;
import org.apache.flume.SourceRunner;
import org.apache.flume.channel.ChannelProcessor;
+import org.apache.flume.channel.ReplicatingChannelSelector;
import org.apache.flume.conf.Configurables;
+import org.apache.flume.conf.channel.ChannelType;
+import org.apache.flume.interceptor.Interceptor;
/**
* The primarily provides helper methods to create and configure the various Flume components.
@@ -37,11 +42,19 @@
*/
public abstract class AbstractFlumeConfiguration {
+ /**
+ * Create amd configure a Channel.
+ * @param name the Channel name.
+ * @param clazz the Channel Class to create.
+ * @param params parameters needed for configuration.
+ * @return The Channel.
+ * @param <T> The specific type of Channel.
+ */
protected <T extends Channel> T configureChannel(
final String name, final Class<T> clazz, final Map<String, String> params) {
T channel;
try {
- channel = clazz.newInstance();
+ channel = clazz.getDeclaredConstructor().newInstance();
} catch (Exception ex) {
throw new FlumeException("Unable to create channel " + name, ex);
}
@@ -50,32 +63,74 @@
return channel;
}
+ /**
+ * Create and configure a Source and its SourcRunner. A ChannelSelector may be provided.
+ * @param name The name of the Source.
+ * @param clazz The Source class to be created and configured.
+ * @param selector The Channel Selector.
+ * @param params parameters required for configuration.
+ * @return The SourceRunner.
+ * @param <T> The Source Class.
+ */
protected <T extends Source> SourceRunner configureSource(
final String name, final Class<T> clazz, final ChannelSelector selector, final Map<String, String> params) {
+ return configureSource(name, clazz, selector, null, params);
+ }
+
+ /**
+ * Create and configure a Source and its SourcRunner. A ChannelSelector and Interceptors may be provided.
+ * @param name The name of the Source.
+ * @param clazz The Source class to be created and configured.
+ * @param selector The Channel Selector.
+ * @param interceptors A List of Interceptors.
+ * @param params parameters required for configuration.
+ * @return The SourceRunner.
+ * @param <T> The Source Class.
+ */
+ protected <T extends Source> SourceRunner configureSource(
+ final String name,
+ final Class<T> clazz,
+ final ChannelSelector selector,
+ final List<Interceptor> interceptors,
+ final Map<String, String> params) {
T source;
try {
- source = clazz.newInstance();
+ source = clazz.getDeclaredConstructor().newInstance();
} catch (Exception ex) {
throw new FlumeException("Unable to create source " + name, ex);
}
source.setName(name);
+ ChannelSelector channelSelector = selector != null ? selector : new ReplicatingChannelSelector();
Configurables.configure(source, createContext(params));
- ChannelProcessor channelProcessor = new ChannelProcessor(selector);
- source.setChannelProcessor(new ChannelProcessor(selector));
+ return configureSource(source, channelSelector);
+ }
+
+ /**
+ * Set up an already configured Source for processing.
+ * @param source The Source.
+ * @param selector The Channel Selector.
+ * @return The SourceRunner.
+ * @param <T> The Source Class.
+ */
+ protected <T extends Source> SourceRunner configureSource(final T source, final ChannelSelector selector) {
+ ChannelSelector channelSelector = selector != null ? selector : new ReplicatingChannelSelector();
+ source.setChannelProcessor(new ChannelProcessor(channelSelector));
return SourceRunner.forSource(source);
}
- protected <T extends Source> SourceRunner configureSource(
- final T source, final ChannelSelector selector, final Map<String, String> params) {
- source.setChannelProcessor(new ChannelProcessor(selector));
- return SourceRunner.forSource(source);
- }
-
+ /**
+ * Set up already configured Sinks for processing.
+ * @param params The SinkProcessor's parameters.
+ * @param clazz The SinkProcessor Class.
+ * @param sinks The list of Sinks.
+ * @return The SinkProcessor.
+ * @param <T> The specific type of the SinkProcessor.
+ */
protected <T extends SinkProcessor> T configureSinkProcessor(
final Map<String, String> params, final Class<T> clazz, final List<Sink> sinks) {
T processor;
try {
- processor = clazz.newInstance();
+ processor = clazz.getDeclaredConstructor().newInstance();
} catch (Exception ex) {
throw new FlumeException("Unable to create SinkProcessor of type: " + clazz.getName(), ex);
}
@@ -84,17 +139,52 @@
return processor;
}
+ /**
+ * Set up the SinkProcessors and SinkRunners for a list of Sinks.
+ * @param processorProperties The SinkProcessor's parameters.
+ * @param sinkProcessorClass The SinkProcessor to create.
+ * @param sinks The Map of Sink Lists with the key being the name of a group of channels to which the sinks are attached.
+ * @return A Map of the SinkRunners
+ */
+ protected Map<String, SinkRunner> createSinkRunners(
+ final Map<String, String> processorProperties,
+ final Class<? extends SinkProcessor> sinkProcessorClass,
+ final Map<String, List<Sink>> sinks) {
+ Map<String, SinkRunner> sinkRunners = new HashMap<>();
+ for (Map.Entry<String, List<Sink>> entry : sinks.entrySet()) {
+ sinkRunners.put(
+ entry.getKey(),
+ createSinkRunner(
+ configureSinkProcessor(processorProperties, sinkProcessorClass, entry.getValue())));
+ }
+ return sinkRunners;
+ }
+
+ /**
+ * Creates the SinkRunner.
+ * @param sinkProcessor The SinkProcessor.
+ * @return A SinkRunner.
+ */
protected SinkRunner createSinkRunner(SinkProcessor sinkProcessor) {
SinkRunner runner = new SinkRunner(sinkProcessor);
runner.setSink(sinkProcessor);
return runner;
}
+ /**
+ * Create and configure a Sink.
+ * @param name The name of the Sink.
+ * @param sinkClazz The Class object for the Sink.
+ * @param channel The Channel attached to the Sink.
+ * @param params Parameters to configure the Sink.
+ * @return The Sink.
+ * @param <T> The type of Sink being created.
+ */
protected <T extends Sink> Sink configureSink(
final String name, final Class<T> sinkClazz, final Channel channel, final Map<String, String> params) {
T sink;
try {
- sink = sinkClazz.newInstance();
+ sink = sinkClazz.getDeclaredConstructor().newInstance();
} catch (Exception ex) {
throw new FlumeException("Unable to create sink " + name, ex);
}
@@ -104,11 +194,18 @@
return sink;
}
+ /**
+ * Create the channel selector and configure it.
+ * @param clazz The Selector class.
+ * @param channels The Channels.
+ * @param params The configuration parameters.
+ * @return The ChannelSelector.
+ */
protected ChannelSelector createChannelSelector(
Class<? extends ChannelSelector> clazz, List<Channel> channels, Map<String, String> params) {
ChannelSelector selector;
try {
- selector = clazz.newInstance();
+ selector = clazz.getDeclaredConstructor().newInstance();
} catch (Exception ex) {
throw new FlumeException("Unable to create channel selector " + clazz.getName(), ex);
}
@@ -117,6 +214,21 @@
return selector;
}
+ @SuppressWarnings("unchecked")
+ protected Class<? extends Channel> getChannelClass(String type) {
+ if (type == null) {
+ return null;
+ } else {
+ ChannelType channelType = null;
+ try {
+ channelType = ChannelType.valueOf(type.toUpperCase(Locale.getDefault()));
+ return Class.forName(channelType.name()).asSubclass(Channel.class);
+ } catch (Exception ex) {
+ return null;
+ }
+ }
+ }
+
/**
* Creates a List from a Varargs array.
*
@@ -129,7 +241,13 @@
return Arrays.asList(items);
}
- private static Context createContext(Map<String, String> map) {
+ /**
+ * Create a Context from the Map.
+ *
+ * @param map contains the configuration parameters for the component being provisioned.
+ * @return The Context.
+ */
+ protected static Context createContext(Map<String, String> map) {
return map != null ? new Context(map) : new Context();
}
}
diff --git a/flume-spring-boot/src/main/java/org/apache/flume/spring/boot/runner/SpringFlume.java b/flume-spring-boot/src/main/java/org/apache/flume/spring/boot/runner/SpringFlume.java
index 3291f26..bd9c929 100644
--- a/flume-spring-boot/src/main/java/org/apache/flume/spring/boot/runner/SpringFlume.java
+++ b/flume-spring-boot/src/main/java/org/apache/flume/spring/boot/runner/SpringFlume.java
@@ -20,6 +20,7 @@
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import java.util.List;
+import org.apache.flume.conf.internal.SuppressFBWarnings;
import org.apache.flume.lifecycle.LifecycleAware;
import org.apache.flume.node.Application;
import org.apache.flume.node.MaterializedConfiguration;
@@ -36,6 +37,7 @@
private final Application application;
@Autowired
+ @SuppressFBWarnings("EI_EXPOSE_REP2")
public SpringFlume(MaterializedConfiguration configuration) {
this.materializedConfiguration = configuration;
List<LifecycleAware> components = Lists.newArrayList();
diff --git a/pom.xml b/pom.xml
index f57611e..a432e7a 100644
--- a/pom.xml
+++ b/pom.xml
@@ -27,7 +27,7 @@
<groupId>org.apache.flume</groupId>
<artifactId>flume-spring-boot-parent</artifactId>
- <version>2.1.0-SNAPSHOT</version>
+ <version>2.0.0-SNAPSHOT</version>
<packaging>pom</packaging>
<name>Flume Spring Boot Parent</name>
@@ -116,6 +116,13 @@
<type>pom</type>
<scope>import</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.flume</groupId>
+ <artifactId>flume-third-party-dependencies</artifactId>
+ <version>${flume.version}</version>
+ <type>pom</type>
+ <scope>import</scope>
+ </dependency>
<!-- Override the versions in this repo -->
<dependency>
<groupId>org.apache.flume</groupId>
