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 20:00:33 UTC
(logging-flume) 01/01: Discover RPC clients through a `RpcClientProvider` SPI
This is an automated email from the ASF dual-hosted git repository.
ppkarwasz pushed a commit to branch feat/rpc-client-provider
in repository https://gitbox.apache.org/repos/asf/logging-flume.git
commit d3cde1928bd7a55bac63b6374e2c9c9f9d83f876
Author: Piotr P. Karwasz <pk...@apache.org>
AuthorDate: Mon Oct 5 20:48:01 2026 +0200
Discover RPC clients through a `RpcClientProvider` SPI
`RpcClientFactory` referred to the Avro and Thrift clients by class name, although they live in other repositories, and created them reflectively before configuring them. Clients are now created by providers discovered with `ServiceLoader` and selected by name through `client.type`; an unregistered provider can still be named by its class. Providers are registered with BND's `@ServiceProvider`, which also generates the JPMS and OSGi metadata.
The types only needed by client implementations move to the new `org.apache.flume.client.spi` package.
Assisted-By: Claude Opus 5.5 <no...@anthropic.com>
---
flume-client/pom.xml | 9 +-
.../org/apache/flume/api/FailoverRpcClient.java | 29 ++-
.../apache/flume/api/LoadBalancingRpcClient.java | 29 ++-
.../flume/api/RpcClientConfigurationConstants.java | 35 ++-
.../org/apache/flume/api/RpcClientFactory.java | 250 ++++++++-------------
.../{api => client/spi}/AbstractRpcClient.java | 14 +-
.../apache/flume/{api => client/spi}/HostInfo.java | 3 +-
.../apache/flume/client/spi/RpcClientProvider.java | 48 ++++
.../spi}/SSLContextAwareAbstractRpcClient.java | 3 +-
.../org/apache/flume/api/TestRpcClientFactory.java | 128 +++++++++++
.../org.apache.flume.client.spi.RpcClientProvider | 1 +
.../org/apache/flume/sink/AbstractRpcSink.java | 2 +-
.../instrumentation/ganglia/GangliaServer.java | 2 +-
flume-parent/pom.xml | 11 +
src/changelog/.2.x.x/add_rpc_client_provider.xml | 7 +
.../modules/ROOT/pages/FlumeDeveloperGuide.adoc | 6 +-
16 files changed, 394 insertions(+), 183 deletions(-)
diff --git a/flume-client/pom.xml b/flume-client/pom.xml
index 122103d7..31bb1d79 100644
--- a/flume-client/pom.xml
+++ b/flume-client/pom.xml
@@ -34,8 +34,8 @@
<!-- 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>26</spotbugs.maxAllowedViolations>
- <pmd.maxAllowedViolations>10</pmd.maxAllowedViolations>
+ <spotbugs.maxAllowedViolations>24</spotbugs.maxAllowedViolations>
+ <pmd.maxAllowedViolations>5</pmd.maxAllowedViolations>
<module.name>org.apache.flume.client</module.name>
</properties>
@@ -46,6 +46,11 @@
<artifactId>log4j-api</artifactId>
</dependency>
+ <dependency>
+ <groupId>biz.aQute.bnd</groupId>
+ <artifactId>biz.aQute.bnd.annotation</artifactId>
+ </dependency>
+
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
diff --git a/flume-client/src/main/java/org/apache/flume/api/FailoverRpcClient.java b/flume-client/src/main/java/org/apache/flume/api/FailoverRpcClient.java
index 7a6fb121..603f1d9f 100644
--- a/flume-client/src/main/java/org/apache/flume/api/FailoverRpcClient.java
+++ b/flume-client/src/main/java/org/apache/flume/api/FailoverRpcClient.java
@@ -16,10 +16,15 @@
*/
package org.apache.flume.api;
+import aQute.bnd.annotation.Resolution;
+import aQute.bnd.annotation.spi.ServiceProvider;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.List;
import java.util.Properties;
+import org.apache.flume.client.spi.AbstractRpcClient;
+import org.apache.flume.client.spi.HostInfo;
+import org.apache.flume.client.spi.RpcClientProvider;
import org.apache.flume.event.Event;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@@ -269,15 +274,35 @@ public class FailoverRpcClient extends AbstractRpcClient implements RpcClient {
}
private void setDefaultProperties(HostInfo hostInfo, Properties props) {
- props.put(RpcClientConfigurationConstants.CONFIG_CLIENT_TYPE, RpcClientFactory.ClientType.DEFAULT.name());
+ props.put(
+ RpcClientConfigurationConstants.CONFIG_CLIENT_TYPE,
+ RpcClientConfigurationConstants.DEFAULT_CLIENT_TYPE);
props.put(RpcClientConfigurationConstants.CONFIG_HOSTS, hostInfo.getReferenceName());
}
- @Override
public void configure(Properties properties) {
configurationProperties = new Properties();
configurationProperties.putAll(properties);
configureHosts(configurationProperties);
}
+
+ /**
+ * Provides the {@value RpcClientConfigurationConstants#FAILOVER_CLIENT_TYPE} client type.
+ */
+ @ServiceProvider(value = RpcClientProvider.class, resolution = Resolution.OPTIONAL)
+ public static final class Provider implements RpcClientProvider {
+
+ @Override
+ public String getName() {
+ return RpcClientConfigurationConstants.FAILOVER_CLIENT_TYPE;
+ }
+
+ @Override
+ public RpcClient create(Properties properties) {
+ FailoverRpcClient client = new FailoverRpcClient();
+ client.configure(properties);
+ return client;
+ }
+ }
}
diff --git a/flume-client/src/main/java/org/apache/flume/api/LoadBalancingRpcClient.java b/flume-client/src/main/java/org/apache/flume/api/LoadBalancingRpcClient.java
index 0b246d31..7cf8c896 100644
--- a/flume-client/src/main/java/org/apache/flume/api/LoadBalancingRpcClient.java
+++ b/flume-client/src/main/java/org/apache/flume/api/LoadBalancingRpcClient.java
@@ -16,12 +16,17 @@
*/
package org.apache.flume.api;
+import aQute.bnd.annotation.Resolution;
+import aQute.bnd.annotation.spi.ServiceProvider;
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.client.spi.AbstractRpcClient;
+import org.apache.flume.client.spi.HostInfo;
+import org.apache.flume.client.spi.RpcClientProvider;
import org.apache.flume.event.Event;
import org.apache.flume.util.OrderSelector;
import org.apache.flume.util.RandomOrderSelector;
@@ -133,7 +138,6 @@ public class LoadBalancingRpcClient extends AbstractRpcClient {
}
}
- @Override
protected void configure(Properties properties) {
clientMap = new HashMap<String, RpcClient>();
configurationProperties = new Properties();
@@ -206,7 +210,9 @@ public class LoadBalancingRpcClient extends AbstractRpcClient {
private Properties getClientConfigurationProperties(String referenceName) {
Properties props = new Properties();
props.putAll(configurationProperties);
- props.put(RpcClientConfigurationConstants.CONFIG_CLIENT_TYPE, RpcClientFactory.ClientType.DEFAULT);
+ props.put(
+ RpcClientConfigurationConstants.CONFIG_CLIENT_TYPE,
+ RpcClientConfigurationConstants.DEFAULT_CLIENT_TYPE);
props.put(RpcClientConfigurationConstants.CONFIG_HOSTS, referenceName);
return props;
@@ -276,4 +282,23 @@ public class LoadBalancingRpcClient extends AbstractRpcClient {
selector.informFailure(failedHost);
}
}
+
+ /**
+ * Provides the {@value RpcClientConfigurationConstants#LOAD_BALANCING_CLIENT_TYPE} client type.
+ */
+ @ServiceProvider(value = RpcClientProvider.class, resolution = Resolution.OPTIONAL)
+ public static final class Provider implements RpcClientProvider {
+
+ @Override
+ public String getName() {
+ return RpcClientConfigurationConstants.LOAD_BALANCING_CLIENT_TYPE;
+ }
+
+ @Override
+ public RpcClient create(Properties properties) {
+ LoadBalancingRpcClient client = new LoadBalancingRpcClient();
+ client.configure(properties);
+ return client;
+ }
+ }
}
diff --git a/flume-client/src/main/java/org/apache/flume/api/RpcClientConfigurationConstants.java b/flume-client/src/main/java/org/apache/flume/api/RpcClientConfigurationConstants.java
index 0c5517ec..bc153378 100644
--- a/flume-client/src/main/java/org/apache/flume/api/RpcClientConfigurationConstants.java
+++ b/flume-client/src/main/java/org/apache/flume/api/RpcClientConfigurationConstants.java
@@ -83,20 +83,39 @@ public final class RpcClientConfigurationConstants {
public static final String CONFIG_MAX_ATTEMPTS = "max-attempts";
/**
- * Configuration key to specify the RpcClient type to be used. The available
- * values are <tt>DEFAULT</tt> which results in the creation of a regular
- * <tt>NettyAvroRpcClient</tt> and <tt>DEFAULT_FAILOVER</tt> which results
- * in the creation of a failover client implementation on top of multiple
- * <tt>NettyAvroRpcClient</tt>s. The default value of this configuration
- * is {@value #DEFAULT_CLIENT_TYPE}.
+ * Configuration key to specify the RpcClient type to be used.
*
+ * <p>The value is the name of a registered {@link org.apache.flume.client.spi.RpcClientProvider},
+ * such as {@value #AVRO_CLIENT_TYPE}, {@value #THRIFT_CLIENT_TYPE}, {@value #FAILOVER_CLIENT_TYPE}
+ * or {@value #LOAD_BALANCING_CLIENT_TYPE}, or the fully qualified class name of a provider.
+ * The default value of this configuration is {@value #DEFAULT_CLIENT_TYPE}.
*/
public static final String CONFIG_CLIENT_TYPE = "client.type";
/**
- * The default client type to be created if no explicit type is specified.
+ * The default client type to be created if no explicit type is specified: an alias of {@value #AVRO_CLIENT_TYPE}.
*/
- public static final String DEFAULT_CLIENT_TYPE = RpcClientFactory.ClientType.DEFAULT.name();
+ public static final String DEFAULT_CLIENT_TYPE = "default";
+
+ /**
+ * Client type of the Avro client, provided by {@code flume-avro-client}.
+ */
+ public static final String AVRO_CLIENT_TYPE = "avro";
+
+ /**
+ * Client type of the Thrift client, provided by {@code flume-thrift-client}.
+ */
+ public static final String THRIFT_CLIENT_TYPE = "thrift";
+
+ /**
+ * Client type of the {@link FailoverRpcClient}.
+ */
+ public static final String FAILOVER_CLIENT_TYPE = "default_failover";
+
+ /**
+ * Client type of the {@link LoadBalancingRpcClient}.
+ */
+ public static final String LOAD_BALANCING_CLIENT_TYPE = "default_loadbalance";
/**
* The selector type used by the <tt>LoadBalancingRpcClient</tt>. This
diff --git a/flume-client/src/main/java/org/apache/flume/api/RpcClientFactory.java b/flume-client/src/main/java/org/apache/flume/api/RpcClientFactory.java
index f15a72ed..2c6bccad 100644
--- a/flume-client/src/main/java/org/apache/flume/api/RpcClientFactory.java
+++ b/flume-client/src/main/java/org/apache/flume/api/RpcClientFactory.java
@@ -16,80 +16,49 @@
*/
package org.apache.flume.api;
+import aQute.bnd.annotation.Cardinality;
+import aQute.bnd.annotation.Resolution;
+import aQute.bnd.annotation.spi.ServiceConsumer;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.io.Reader;
import java.lang.reflect.InvocationTargetException;
import java.util.Locale;
+import java.util.Map;
import java.util.Properties;
+import java.util.ServiceLoader;
+import java.util.TreeMap;
+import org.apache.flume.client.spi.RpcClientProvider;
/**
- * Factory class to construct Flume {@link RPCClient} implementations.
+ * Factory class to construct Flume {@link RpcClient} implementations.
+ *
+ * <p>The {@value RpcClientConfigurationConstants#CONFIG_CLIENT_TYPE} property selects the client:
+ * either the name of a provider registered through {@link java.util.ServiceLoader},
+ * or the fully qualified class name of an unregistered {@link RpcClientProvider}.
*/
+@ServiceConsumer(value = RpcClientProvider.class, resolution = Resolution.OPTIONAL, cardinality = Cardinality.MULTIPLE)
public class RpcClientFactory {
- private static final String AVRO_CLASS_NAME = "org.apache.flume.client.avro.NettyAvroRpcClient";
- private static final String THRIFT_CLASS_NAME = "org.apache.flume.client.thrift.ThriftRpcClient";
- private static final String NEW_INSTANCE = "newInstance";
+ private RpcClientFactory() {}
/**
- * Returns an instance of {@link RpcClient}, optionally with failover.
- * To create a failover client, the properties object should have a
- * property <tt>client.type</tt> which has the value "failover". The client
- * connects to hosts specified by <tt>hosts</tt> property in given properties.
- *
- * @see FailoverRpcClient
- * <p>
- * If no <tt>client.type</tt> is specified, a default client that connects to
- * single host at a given port is created.(<tt>type</tt> can also simply be
- * <tt>DEFAULT</tt> for the default client).
+ * Returns an instance of {@link RpcClient} configured with the given properties.
*
- * @see org.apache.flume.api.NettyAvroClient
+ * <p>If {@value RpcClientConfigurationConstants#CONFIG_CLIENT_TYPE} is not specified,
+ * a client of type {@value RpcClientConfigurationConstants#DEFAULT_CLIENT_TYPE} is created.
*
* @param properties The properties to instantiate the client with.
+ * @throws IllegalArgumentException if the client type is unknown or the properties are invalid.
* @throws IOException if the client fails to connect.
*/
- @SuppressWarnings("unchecked")
public static RpcClient getInstance(Properties properties) throws IOException {
- String type = null;
- type = properties.getProperty(RpcClientConfigurationConstants.CONFIG_CLIENT_TYPE);
+ String type = properties.getProperty(RpcClientConfigurationConstants.CONFIG_CLIENT_TYPE);
if (type == null || type.isEmpty()) {
- type = ClientType.DEFAULT.name();
- }
- ClientType clientType = null;
- try {
- clientType = ClientType.valueOf(type.toUpperCase(Locale.ENGLISH));
- } catch (IllegalArgumentException e) {
- clientType = ClientType.OTHER;
- }
- AbstractRpcClient client;
- if (clientType.isNewInstance()) {
- return createClient(clientType.clientClassName, properties);
- }
- Class<? extends AbstractRpcClient> clazz;
- try {
- String clientClassType = type;
-
- if (!clientType.equals(ClientType.OTHER)) {
- clientClassType = clientType.getClientClassName();
- }
- clazz = (Class<? extends AbstractRpcClient>) Class.forName(clientClassType);
- } catch (ClassNotFoundException e) {
- throw new IllegalArgumentException("No such client: " + type, e);
+ type = RpcClientConfigurationConstants.DEFAULT_CLIENT_TYPE;
}
-
- try {
- client = clazz.newInstance();
- } catch (InstantiationException e) {
- throw new IllegalArgumentException(
- "Cannot instantiate client " + clazz.getCanonicalName() + " Exception follows:", e);
- } catch (IllegalAccessException e) {
- throw new IllegalArgumentException(
- "Cannot instantiate client " + clazz.getCanonicalName() + "Exception follows:", e);
- }
- client.configure(properties);
- return client;
+ return getProvider(type).create(properties);
}
/**
@@ -97,13 +66,13 @@ public class RpcClientFactory {
* to a {@link Properties} file.
* @param propertiesFile Valid properties file
* @return RpcClient configured according to the given Properties file.
- * @throws FileNotFoundException If the file cannot be found
- * @throws IOException If there is an IO error
+ * @throws IOException If the file cannot be read or the client fails to connect
*/
public static RpcClient getInstance(File propertiesFile) throws IOException {
- Reader reader = new FileReader(propertiesFile);
Properties props = new Properties();
- props.load(reader);
+ try (Reader reader = new FileReader(propertiesFile)) {
+ props.load(reader);
+ }
return getInstance(props);
}
@@ -145,35 +114,40 @@ public class RpcClientFactory {
* @throws IOException if the client fails to connect.
*/
public static RpcClient getDefaultInstance(String hostname, Integer port, Integer batchSize) throws IOException {
+ return getProvider(RpcClientConfigurationConstants.DEFAULT_CLIENT_TYPE)
+ .create(singleHostProperties(hostname, port, batchSize));
+ }
- if (hostname == null) {
- throw new NullPointerException("hostname must not be null");
- }
- if (port == null) {
- throw new NullPointerException("port must not be null");
- }
- if (batchSize == null) {
- throw new NullPointerException("batchSize must not be null");
- }
+ /**
+ * Returns an instance of {@link RpcClient} connected to the specified
+ * {@code hostname} and {@code port} using Thrift.
+ * @throws IOException if the client fails to connect.
+ */
+ public static RpcClient getThriftInstance(String hostname, Integer port, Integer batchSize) throws IOException {
+ return getProvider(RpcClientConfigurationConstants.THRIFT_CLIENT_TYPE)
+ .create(singleHostProperties(hostname, port, batchSize));
+ }
- Properties props = new Properties();
- props.setProperty(RpcClientConfigurationConstants.CONFIG_HOSTS, "h1");
- props.setProperty(RpcClientConfigurationConstants.CONFIG_HOSTS_PREFIX + "h1", hostname + ":" + port.intValue());
- props.setProperty(RpcClientConfigurationConstants.CONFIG_BATCH_SIZE, batchSize.toString());
- return createClient(AVRO_CLASS_NAME, props);
+ /**
+ * Returns an instance of {@link RpcClient} connected to the specified
+ * {@code hostname} and {@code port} using Thrift.
+ * @throws IOException if the client fails to connect.
+ */
+ public static RpcClient getThriftInstance(String hostname, Integer port) throws IOException {
+ return getThriftInstance(hostname, port, RpcClientConfigurationConstants.DEFAULT_BATCH_SIZE);
}
/**
- * Return an {@linkplain RpcClient} that uses Thrift for communicating with
- * the next hop. The next hop must have a ThriftSource listening on the
- * specified port.
- * @param hostname - The hostname of the next hop.
- * @param port - The port on which the ThriftSource is listening
- * @param batchSize - batch size of each transaction.
- * @return an {@linkplain RpcClient} which uses thrift configured with the
- * given parameters.
+ * Returns an instance of {@link RpcClient} configured with the given properties using Thrift.
+ * @throws IOException if the client fails to connect.
*/
- public static RpcClient getThriftInstance(String hostname, Integer port, Integer batchSize) throws IOException {
+ public static RpcClient getThriftInstance(Properties props) throws IOException {
+ props.setProperty(
+ RpcClientConfigurationConstants.CONFIG_CLIENT_TYPE, RpcClientConfigurationConstants.THRIFT_CLIENT_TYPE);
+ return getInstance(props);
+ }
+
+ private static Properties singleHostProperties(String hostname, Integer port, Integer batchSize) {
if (hostname == null) {
throw new NullPointerException("hostname must not be null");
}
@@ -183,92 +157,62 @@ public class RpcClientFactory {
if (batchSize == null) {
throw new NullPointerException("batchSize must not be null");
}
-
Properties props = new Properties();
props.setProperty(RpcClientConfigurationConstants.CONFIG_HOSTS, "h1");
props.setProperty(RpcClientConfigurationConstants.CONFIG_HOSTS_PREFIX + "h1", hostname + ":" + port.intValue());
props.setProperty(RpcClientConfigurationConstants.CONFIG_BATCH_SIZE, batchSize.toString());
- return createClient(THRIFT_CLASS_NAME, props);
- }
-
- private static RpcClient createClient(String className, Properties props) throws IOException {
- try {
- @SuppressWarnings("unchecked")
- Class<? extends AbstractRpcClient> clazz = (Class<? extends AbstractRpcClient>) Class.forName(className);
- AbstractRpcClient client =
- (AbstractRpcClient) clazz.getMethod(NEW_INSTANCE).invoke(null);
- client.configure(props);
- return client;
- } catch (ClassNotFoundException
- | NoSuchMethodException
- | IllegalAccessException
- | InvocationTargetException e) {
- throw new IllegalStateException(
- "Cannot instantiate client. Implementation " + className + " failed to load.", e);
- }
+ return props;
}
/**
- * Return an {@linkplain RpcClient} that uses Thrift for communicating with
- * the next hop. The next hop must have a ThriftSource listening on the
- * specified port. This will use the default batch size. See {@linkplain
- * RpcClientConfigurationConstants}
- * @param hostname - The hostname of the next hop.
- * @param port - The port on which the ThriftSource is listening
- * @return - An {@linkplain RpcClient} which uses thrift configured with the
- * given parameters.
- */
- public static RpcClient getThriftInstance(String hostname, Integer port) throws IOException {
- return getThriftInstance(hostname, port, RpcClientConfigurationConstants.DEFAULT_BATCH_SIZE);
- }
-
- /**
- * Return an {@linkplain RpcClient} that uses Thrift for communicating with
- * the next hop.
- * @param props
- * @return - An {@linkplain RpcClient} which uses thrift configured with the
- * given parameters.
+ * Returns the provider for a client type.
+ *
+ * <p>Registered providers are matched by name, ignoring case;
+ * {@value RpcClientConfigurationConstants#DEFAULT_CLIENT_TYPE} is an alias of
+ * {@value RpcClientConfigurationConstants#AVRO_CLIENT_TYPE}.
+ * Otherwise, the type is the fully qualified class name of a provider.
*/
- public static RpcClient getThriftInstance(Properties props) throws IOException {
- props.setProperty(RpcClientConfigurationConstants.CONFIG_CLIENT_TYPE, ClientType.THRIFT.clientClassName);
- return getInstance(props);
- }
-
- public static enum ClientType {
- OTHER(null),
- DEFAULT(AVRO_CLASS_NAME, true),
- DEFAULT_FAILOVER(FailoverRpcClient.class.getCanonicalName()),
- DEFAULT_LOADBALANCE(LoadBalancingRpcClient.class.getCanonicalName()),
- THRIFT(THRIFT_CLASS_NAME, true);
-
- private final String clientClassName;
- private final boolean newInstance;
-
- private ClientType(String className) {
- this.clientClassName = className;
- this.newInstance = false;
- }
-
- private ClientType(String className, boolean newInstance) {
- this.clientClassName = className;
- this.newInstance = newInstance;
+ static RpcClientProvider getProvider(String type) {
+ String name = type.equalsIgnoreCase(RpcClientConfigurationConstants.DEFAULT_CLIENT_TYPE)
+ ? RpcClientConfigurationConstants.AVRO_CLIENT_TYPE
+ : type;
+ Map<String, RpcClientProvider> providers = loadProviders();
+ RpcClientProvider provider = providers.get(name.toLowerCase(Locale.ROOT));
+ if (provider != null) {
+ return provider;
}
-
- private static ClientType getClientType(String className) {
- for (ClientType type : ClientType.values()) {
- if (type.clientClassName.equals(className)) {
- return type;
- }
+ try {
+ Class<?> clazz = Class.forName(type, true, RpcClientFactory.class.getClassLoader());
+ if (!RpcClientProvider.class.isAssignableFrom(clazz)) {
+ throw new IllegalArgumentException(
+ "Client type " + type + " does not implement " + RpcClientProvider.class.getName());
}
- return OTHER;
- }
-
- protected String getClientClassName() {
- return this.clientClassName;
+ return (RpcClientProvider) clazz.getConstructor().newInstance();
+ } catch (ClassNotFoundException e) {
+ throw new IllegalArgumentException(
+ "Unknown client type " + type + ": add the artifact that provides it, or use one of "
+ + providers.keySet(),
+ e);
+ } catch (InstantiationException
+ | IllegalAccessException
+ | InvocationTargetException
+ | NoSuchMethodException e) {
+ throw new IllegalArgumentException("Cannot instantiate client provider " + type, e);
}
+ }
- protected boolean isNewInstance() {
- return this.newInstance;
+ private static Map<String, RpcClientProvider> loadProviders() {
+ Map<String, RpcClientProvider> providers = new TreeMap<>();
+ for (RpcClientProvider provider :
+ ServiceLoader.load(RpcClientProvider.class, RpcClientFactory.class.getClassLoader())) {
+ String name = provider.getName().toLowerCase(Locale.ROOT);
+ RpcClientProvider previous = providers.putIfAbsent(name, provider);
+ if (previous != null) {
+ throw new IllegalStateException("Client type " + name + " is provided by both "
+ + previous.getClass().getName() + " and "
+ + provider.getClass().getName());
+ }
}
+ return providers;
}
}
diff --git a/flume-client/src/main/java/org/apache/flume/api/AbstractRpcClient.java b/flume-client/src/main/java/org/apache/flume/client/spi/AbstractRpcClient.java
similarity index 87%
rename from flume-client/src/main/java/org/apache/flume/api/AbstractRpcClient.java
rename to flume-client/src/main/java/org/apache/flume/client/spi/AbstractRpcClient.java
index 6e2f04f5..6dd68059 100644
--- a/flume-client/src/main/java/org/apache/flume/api/AbstractRpcClient.java
+++ b/flume-client/src/main/java/org/apache/flume/client/spi/AbstractRpcClient.java
@@ -14,11 +14,14 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.flume.api;
+package org.apache.flume.client.spi;
import java.io.IOException;
import java.util.List;
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.event.Event;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@@ -47,15 +50,6 @@ public abstract class AbstractRpcClient implements RpcClient {
@Override
public abstract void close() throws IOException;
- /**
- * Configure the client using the given properties object.
- * @param properties
- * @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 IOException;
-
/**
* This is to parse the batch size config for rpc clients
* @param properties config
diff --git a/flume-client/src/main/java/org/apache/flume/api/HostInfo.java b/flume-client/src/main/java/org/apache/flume/client/spi/HostInfo.java
similarity index 97%
rename from flume-client/src/main/java/org/apache/flume/api/HostInfo.java
rename to flume-client/src/main/java/org/apache/flume/client/spi/HostInfo.java
index b1006ee3..9007464a 100644
--- a/flume-client/src/main/java/org/apache/flume/api/HostInfo.java
+++ b/flume-client/src/main/java/org/apache/flume/client/spi/HostInfo.java
@@ -14,11 +14,12 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.flume.api;
+package org.apache.flume.client.spi;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
+import org.apache.flume.api.RpcClientConfigurationConstants;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
diff --git a/flume-client/src/main/java/org/apache/flume/client/spi/RpcClientProvider.java b/flume-client/src/main/java/org/apache/flume/client/spi/RpcClientProvider.java
new file mode 100644
index 00000000..54f92ac7
--- /dev/null
+++ b/flume-client/src/main/java/org/apache/flume/client/spi/RpcClientProvider.java
@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to you under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flume.client.spi;
+
+import java.io.IOException;
+import java.util.Properties;
+import org.apache.flume.api.RpcClient;
+
+/**
+ * Creates the {@link RpcClient} instances of one client type.
+ *
+ * <p>Providers are discovered with {@link java.util.ServiceLoader}
+ * and selected by the {@code client.type} configuration property:
+ * register implementations in {@code META-INF/services/org.apache.flume.client.spi.RpcClientProvider}.
+ */
+public interface RpcClientProvider {
+
+ /**
+ * Returns the value of {@code client.type} that selects this provider.
+ *
+ * <p>Client types are compared case-insensitively.
+ */
+ String getName();
+
+ /**
+ * Creates a client configured with the given properties.
+ *
+ * @param properties the client configuration.
+ * @return a new client, ready-to-send events.
+ * @throws IllegalArgumentException if the configuration is invalid.
+ * @throws IOException if the client fails to connect.
+ */
+ RpcClient create(Properties properties) throws IOException;
+}
diff --git a/flume-client/src/main/java/org/apache/flume/api/SSLContextAwareAbstractRpcClient.java b/flume-client/src/main/java/org/apache/flume/client/spi/SSLContextAwareAbstractRpcClient.java
similarity index 97%
rename from flume-client/src/main/java/org/apache/flume/api/SSLContextAwareAbstractRpcClient.java
rename to flume-client/src/main/java/org/apache/flume/client/spi/SSLContextAwareAbstractRpcClient.java
index 7ffdb77b..9070e2b0 100644
--- a/flume-client/src/main/java/org/apache/flume/api/SSLContextAwareAbstractRpcClient.java
+++ b/flume-client/src/main/java/org/apache/flume/client/spi/SSLContextAwareAbstractRpcClient.java
@@ -14,13 +14,14 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.flume.api;
+package org.apache.flume.client.spi;
import java.util.Arrays;
import java.util.LinkedHashSet;
import java.util.Objects;
import java.util.Properties;
import java.util.Set;
+import org.apache.flume.api.RpcClientConfigurationConstants;
import org.apache.flume.util.SSLUtil;
public abstract class SSLContextAwareAbstractRpcClient extends AbstractRpcClient {
diff --git a/flume-client/src/test/java/org/apache/flume/api/TestRpcClientFactory.java b/flume-client/src/test/java/org/apache/flume/api/TestRpcClientFactory.java
new file mode 100644
index 00000000..3db4acdd
--- /dev/null
+++ b/flume-client/src/test/java/org/apache/flume/api/TestRpcClientFactory.java
@@ -0,0 +1,128 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to you under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flume.api;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+
+import java.util.List;
+import java.util.Properties;
+import org.apache.flume.client.spi.RpcClientProvider;
+import org.apache.flume.event.Event;
+import org.junit.Test;
+
+public class TestRpcClientFactory {
+
+ @Test
+ public void testDefaultTypeIsAvro() throws Exception {
+ assertSame(
+ FakeAvroProvider.class, RpcClientFactory.getProvider("default").getClass());
+ RpcClient client = RpcClientFactory.getInstance(new Properties());
+ assertEquals("avro", ((FakeClient) client).type);
+ }
+
+ @Test
+ public void testTypesAreCaseInsensitive() {
+ assertSame(FakeAvroProvider.class, RpcClientFactory.getProvider("AvRo").getClass());
+ assertSame(
+ FailoverRpcClient.Provider.class,
+ RpcClientFactory.getProvider("DEFAULT_FAILOVER").getClass());
+ assertSame(
+ LoadBalancingRpcClient.Provider.class,
+ RpcClientFactory.getProvider("default_loadbalance").getClass());
+ }
+
+ @Test
+ public void testUnregisteredProviderByClassName() throws Exception {
+ Properties properties = new Properties();
+ properties.setProperty(
+ RpcClientConfigurationConstants.CONFIG_CLIENT_TYPE, UnregisteredProvider.class.getName());
+ RpcClient client = RpcClientFactory.getInstance(properties);
+ assertEquals("unregistered", ((FakeClient) client).type);
+ }
+
+ @Test
+ public void testUnknownType() {
+ IllegalArgumentException e =
+ assertThrows(IllegalArgumentException.class, () -> RpcClientFactory.getProvider("carrier-pigeon"));
+ assertTrue(e.getMessage(), e.getMessage().contains("avro"));
+ assertTrue(e.getMessage(), e.getMessage().contains("default_failover"));
+ }
+
+ @Test
+ public void testClassThatIsNotAProvider() {
+ assertThrows(IllegalArgumentException.class, () -> RpcClientFactory.getProvider(String.class.getName()));
+ }
+
+ /** Stands in for the provider of {@code flume-avro-client}. */
+ public static final class FakeAvroProvider implements RpcClientProvider {
+
+ @Override
+ public String getName() {
+ return "AVRO";
+ }
+
+ @Override
+ public RpcClient create(Properties properties) {
+ return new FakeClient("avro");
+ }
+ }
+
+ /** A provider that is not registered with the {@link java.util.ServiceLoader}. */
+ public static final class UnregisteredProvider implements RpcClientProvider {
+
+ @Override
+ public String getName() {
+ return "unregistered";
+ }
+
+ @Override
+ public RpcClient create(Properties properties) {
+ return new FakeClient("unregistered");
+ }
+ }
+
+ private static final class FakeClient implements RpcClient {
+
+ private final String type;
+
+ private FakeClient(String type) {
+ this.type = type;
+ }
+
+ @Override
+ public int getBatchSize() {
+ return 1;
+ }
+
+ @Override
+ public void append(Event event) {}
+
+ @Override
+ public void appendBatch(List<Event> events) {}
+
+ @Override
+ public boolean isActive() {
+ return true;
+ }
+
+ @Override
+ public void close() {}
+ }
+}
diff --git a/flume-client/src/test/resources/META-INF/services/org.apache.flume.client.spi.RpcClientProvider b/flume-client/src/test/resources/META-INF/services/org.apache.flume.client.spi.RpcClientProvider
new file mode 100644
index 00000000..bad019aa
--- /dev/null
+++ b/flume-client/src/test/resources/META-INF/services/org.apache.flume.client.spi.RpcClientProvider
@@ -0,0 +1 @@
+org.apache.flume.api.TestRpcClientFactory$FakeAvroProvider
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 a0c49797..58e2988a 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
@@ -33,9 +33,9 @@ import org.apache.flume.Context;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.FlumeException;
import org.apache.flume.Transaction;
-import org.apache.flume.api.AbstractRpcClient;
import org.apache.flume.api.RpcClient;
import org.apache.flume.api.RpcClientConfigurationConstants;
+import org.apache.flume.client.spi.AbstractRpcClient;
import org.apache.flume.conf.BatchSizeSupported;
import org.apache.flume.conf.Configurable;
import org.apache.flume.event.Event;
diff --git a/flume-ng-instrumentation/flume-ganglia-monitor/src/main/java/org/apache/flume/instrumentation/ganglia/GangliaServer.java b/flume-ng-instrumentation/flume-ganglia-monitor/src/main/java/org/apache/flume/instrumentation/ganglia/GangliaServer.java
index 94724246..c60b9bee 100644
--- a/flume-ng-instrumentation/flume-ganglia-monitor/src/main/java/org/apache/flume/instrumentation/ganglia/GangliaServer.java
+++ b/flume-ng-instrumentation/flume-ganglia-monitor/src/main/java/org/apache/flume/instrumentation/ganglia/GangliaServer.java
@@ -30,7 +30,7 @@ import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import org.apache.flume.Context;
import org.apache.flume.FlumeException;
-import org.apache.flume.api.HostInfo;
+import org.apache.flume.client.spi.HostInfo;
import org.apache.flume.conf.ConfigurationException;
import org.apache.flume.instrumentation.MonitorService;
import org.apache.flume.instrumentation.util.JMXPollUtil;
diff --git a/flume-parent/pom.xml b/flume-parent/pom.xml
index cca354cc..f1a13eca 100644
--- a/flume-parent/pom.xml
+++ b/flume-parent/pom.xml
@@ -94,6 +94,9 @@
Implementation-Timestamp: ${project.build.outputTimestamp}
Purl: pkg:maven/${project.groupId}/${project.artifactId}@${project.version}</bnd-extra-config>
+ <!-- Generate `requires static` for the annotation libraries, which are only needed at compile time.
+ Overrides the `bnd-jpms-module-info` of `logging-parent`, hence repeats `$[bnd-module-name];access=0`. -->
+ <bnd-jpms-module-info>$[bnd-module-name];access=0;modules="biz.aQute.bnd.annotation"</bnd-jpms-module-info>
<!-- Set default encoding to UTF-8 to remove maven complaints -->
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
@@ -137,6 +140,14 @@
<scope>import</scope>
</dependency>
+ <!-- BND annotations, e.g. `@ServiceProvider` to generate `META-INF/services` files -->
+ <dependency>
+ <groupId>biz.aQute.bnd</groupId>
+ <artifactId>biz.aQute.bnd.annotation</artifactId>
+ <version>${bnd-maven-plugin.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
<dependency>
<groupId>org.awaitility</groupId>
<artifactId>awaitility</artifactId>
diff --git a/src/changelog/.2.x.x/add_rpc_client_provider.xml b/src/changelog/.2.x.x/add_rpc_client_provider.xml
new file mode 100644
index 00000000..447e2005
--- /dev/null
+++ b/src/changelog/.2.x.x/add_rpc_client_provider.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">Discover RPC client implementations through the new `RpcClientProvider` service provider interface instead of hard-coded class names.</description>
+</entry>
diff --git a/src/site/antora/modules/ROOT/pages/FlumeDeveloperGuide.adoc b/src/site/antora/modules/ROOT/pages/FlumeDeveloperGuide.adoc
index 48b91de5..28a0ec64 100644
--- a/src/site/antora/modules/ROOT/pages/FlumeDeveloperGuide.adoc
+++ b/src/site/antora/modules/ROOT/pages/FlumeDeveloperGuide.adoc
@@ -254,11 +254,13 @@ a1.sinks.k1.type = logger
For more flexibility, the default Flume client implementations
(`NettyAvroRpcClient` and `ThriftRpcClient`) can be configured with these
-properties:
+properties.
+The `avro` client type requires `flume-avro-client` on the classpath,
+and the `thrift` client type requires `flume-thrift-client`.
[source,properties]
----
-client.type = default (for avro) or thrift (for thrift)
+client.type = avro (or its alias default) or thrift
hosts = h1 # default client accepts only 1 host
# (additional hosts will be ignored)