MINOR: Refactor abstractConfig#configuredInstance
Reviewers: Jason Gustafson
@@ -355,6 +355,29 @@ public class AbstractConfig {
log.warn("The configuration '{}' was supplied but isn't a known config.", key);
}
+ private <T> T getConfiguredInstance(Object klass, Class<T> t, Map<String, Object> configPairs) {
+ if (klass == null)
+ return null;
+
+ Object o;
+ if (klass instanceof String) {
+ try {
+ o = Utils.newInstance((String) klass, t);
+ } catch (ClassNotFoundException e) {
+ throw new KafkaException("Class " + klass + " cannot be found", e);
+ }
+ } else if (klass instanceof Class<?>) {
+ o = Utils.newInstance((Class<?>) klass);
+ } else
+ throw new KafkaException("Unexpected element of type " + klass.getClass().getName() + ", expected String or Class");
+ if (!t.isInstance(o))
+ throw new KafkaException(klass + " is not an instance of " + t.getName());
+ if (o instanceof Configurable)
+ ((Configurable) o).configure(configPairs);
+
+ return t.cast(o);
+ }
+
/**
* Get a configured instance of the give class specified by the given configuration key. If the object implements
* Configurable configure it using the configuration.
@@ -365,14 +388,8 @@ public class AbstractConfig {
*/
public <T> T getConfiguredInstance(String key, Class<T> t) {
Class<?> c = getClass(key);
- if (c == null)
- return null;
- Object o = Utils.newInstance(c);
- if (!t.isInstance(o))
- throw new KafkaException(c.getName() + " is not an instance of " + t.getName());
- if (o instanceof Configurable)
- ((Configurable) o).configure(originals());
- return t.cast(o);
+
+ return getConfiguredInstance(c, t, originals());
}
/**
@@ -400,7 +417,6 @@ public class AbstractConfig {
return getConfiguredInstances(getList(key), t, configOverrides);
}
-
/**
* Get a list of configured instances of the given class specified by the given configuration key. The configuration
* may specify either null or an empty string to indicate no configured instances. In both cases, this method
@@ -417,21 +433,7 @@ public class AbstractConfig {
Map<String, Object> configPairs = originals();
configPairs.putAll(configOverrides);
for (Object klass : classNames) {
- Object o;
- if (klass instanceof String) {
- try {
- o = Utils.newInstance((String) klass, t);
- } catch (ClassNotFoundException e) {
- throw new KafkaException(klass + " ClassNotFoundException exception occurred", e);
- }
- } else if (klass instanceof Class<?>) {
- o = Utils.newInstance((Class<?>) klass);
- } else
- throw new KafkaException("List contains element of type " + klass.getClass().getName() + ", expected String or Class");
- if (!t.isInstance(o))
- throw new KafkaException(klass + " is not an instance of " + t.getName());
- if (o instanceof Configurable)
- ((Configurable) o).configure(configPairs);
+ Object o = getConfiguredInstance(klass, t, configPairs);
objects.add(t.cast(o));
}
return objects;
