root
/
kafka
mirror of https://github.com/apache/kafka.git
1
0
Fork 0
Code Issues Actions 5 Packages Projects Releases Wiki Activity

Remove Copycat reflection utils, use existing Utils and ConfigDef functionality from clients package.

This commit is contained in:
Ewen Cheslack-Postava 2015-07-31 21:34:32 -07:00
parent be5c387a0e
commit e3451427f2
4 changed files with 33 additions and 92 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
clients/src/main/java/org/apache/kafka/common/utils/Utils.java
View File

@ -271,7 +271,7 @@ public class Utils {
/**
* Instantiate the class
*/
public static Object newInstance(Class<?> c) {
public static <T> T newInstance(Class<T> c) {
try {
return c.newInstance();
} catch (IllegalAccessException e) {
@ -281,6 +281,17 @@ public class Utils {
}
}
/**
* Look up the class by name and instantiate it.
* @param klass class name
* @param base super class of the class to be instantiated
* @param <T>
* @return the new instance
*/
public static <T> T newInstance(String klass, Class<T> base) throws ClassNotFoundException {
return Utils.newInstance(Class.forName(klass).asSubclass(base));
}
/**
* Generates 32 bit murmur2 hash from byte array
* @param data byte array to hash

25
copycat-runtime/src/main/java/org/apache/kafka/copycat/runtime/Worker.java
View File

@ -32,7 +32,6 @@ import org.apache.kafka.copycat.sink.SinkTask;
import org.apache.kafka.copycat.source.SourceTask;
import org.apache.kafka.copycat.storage.*;
import org.apache.kafka.copycat.util.ConnectorTaskId;
import org.apache.kafka.copycat.util.Reflection;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -66,9 +65,7 @@ public class Worker {
public Worker(WorkerConfig config) {
this(new SystemTime(), config,
Reflection.instantiateConfigurable(
config.getClass(WorkerConfig.OFFSET_STORAGE_CLASS_CONFIG).getName(),
OffsetBackingStore.class, config.getUnusedProperties()),
config.getConfiguredInstance(WorkerConfig.OFFSET_STORAGE_CLASS_CONFIG, OffsetBackingStore.class),
null, null, null, null);
}
@ -77,38 +74,34 @@ public class Worker {
Deserializer offsetKeyDeserializer, Deserializer offsetValueDeserializer) {
this.time = time;
this.config = config;
this.converter = Reflection.instantiate(config.getClass(WorkerConfig.CONVERTER_CLASS_CONFIG).getName(), Converter.class);
this.converter = config.getConfiguredInstance(WorkerConfig.CONVERTER_CLASS_CONFIG, Converter.class);
this.offsetBackingStore = offsetBackingStore;
if (offsetKeySerializer != null) {
this.offsetKeySerializer = offsetKeySerializer;
} else {
this.offsetKeySerializer = Reflection.instantiate(
config.getClass(WorkerConfig.KEY_SERIALIZER_CLASS_CONFIG).getName(), Serializer.class);
this.offsetKeySerializer = config.getConfiguredInstance(WorkerConfig.KEY_SERIALIZER_CLASS_CONFIG, Serializer.class);
this.offsetKeySerializer.configure(config.getOriginalProperties(), true);
}
if (offsetValueSerializer != null) {
this.offsetValueSerializer = offsetValueSerializer;
} else {
this.offsetValueSerializer = Reflection.instantiate(
config.getClass(WorkerConfig.VALUE_SERIALIZER_CLASS_CONFIG).getName(), Serializer.class);
this.offsetValueSerializer = config.getConfiguredInstance(WorkerConfig.VALUE_SERIALIZER_CLASS_CONFIG, Serializer.class);
this.offsetValueSerializer.configure(config.getOriginalProperties(), false);
}
if (offsetKeyDeserializer != null) {
this.offsetKeyDeserializer = offsetKeyDeserializer;
} else {
this.offsetKeyDeserializer = Reflection.instantiate(
config.getClass(WorkerConfig.KEY_DESERIALIZER_CLASS_CONFIG).getName(), Deserializer.class);
this.offsetKeyDeserializer = config.getConfiguredInstance(WorkerConfig.KEY_DESERIALIZER_CLASS_CONFIG, Deserializer.class);
this.offsetKeyDeserializer.configure(config.getOriginalProperties(), true);
}
if (offsetValueDeserializer != null) {
this.offsetValueDeserializer = offsetValueDeserializer;
} else {
this.offsetValueDeserializer = Reflection.instantiate(
config.getClass(WorkerConfig.VALUE_DESERIALIZER_CLASS_CONFIG).getName(), Deserializer.class);
this.offsetValueDeserializer = config.getConfiguredInstance(WorkerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, Deserializer.class);
this.offsetValueDeserializer.configure(config.getOriginalProperties(), false);
}
}
@ -213,7 +206,11 @@ public class Worker {
}
private static Task instantiateTask(String taskClassName) {
return Reflection.instantiate(taskClassName, Task.class);
try {
return Utils.newInstance(Class.forName(taskClassName).asSubclass(Task.class));
} catch (ClassNotFoundException e) {
throw new CopycatRuntimeException("Task class not found", e);
}
}
public void stopTask(ConnectorTaskId id) throws CopycatException {

12
copycat-runtime/src/main/java/org/apache/kafka/copycat/runtime/standalone/StandaloneCoordinator.java
View File

@ -55,7 +55,11 @@ public class StandaloneCoordinator implements Coordinator {
String storage = configs.getProperty(STORAGE_CONFIG);
if (storage != null && !storage.isEmpty()) {
configStorage = Reflection.instantiate(storage, ConfigStorage.class);
try {
configStorage = Utils.newInstance(storage, ConfigStorage.class);
} catch (ClassNotFoundException e) {
throw new CopycatRuntimeException("Couldn't configure storage", e);
}
configStorage.configure(configs);
} else {
configStorage = null;
@ -153,7 +157,11 @@ public class StandaloneCoordinator implements Coordinator {
}
private static Connector instantiateConnector(String className) {
return Reflection.instantiate(className, Connector.class);
try {
return Utils.newInstance(className, Connector.class);
} catch (ClassNotFoundException e) {
throw new CopycatRuntimeException("Couldn't instantiate connector class", e);
}
}
private void destroyConnector(String connName) {

75
copycat-runtime/src/main/java/org/apache/kafka/copycat/util/Reflection.java
View File

@ -1,75 +0,0 @@
/**
* 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
* <p/>
* http://www.apache.org/licenses/LICENSE-2.0
* <p/>
* 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.kafka.copycat.util;
import org.apache.kafka.copycat.errors.CopycatRuntimeException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.lang.reflect.InvocationTargetException;
import java.util.Properties;
public class Reflection {
private static final Logger log = LoggerFactory.getLogger(Reflection.class);
/**
* Instantiate the class, which is a subclass of the specified type.
* @param className fully qualified name of the class to instantiate
* @param superType required supertype of the class
* @param <T>
* @return a new instance of the class instantiated with the default constructor
*/
public static <T> T instantiate(String className, Class<T> superType) {
try {
Class<? extends T> taskClass = Class.forName(className).asSubclass(superType);
return taskClass.getConstructor().newInstance();
} catch (ClassNotFoundException e) {
log.error("Class not found: " + className, e);
throw new CopycatRuntimeException(e);
} catch (ClassCastException e) {
log.error("Specified class " + className + " is not a subclass of " + superType + ":", e);
throw new CopycatRuntimeException(e);
} catch (NoSuchMethodException e) {
log.error("Class does not have a default constructor: " + className, e);
throw new CopycatRuntimeException(e);
} catch (InstantiationException e) {
log.error("Class couldn't be instantiated: " + className, e);
throw new CopycatRuntimeException(e);
} catch (InvocationTargetException e) {
log.error("Class constructor threw an exception: " + className, e);
throw new CopycatRuntimeException(e);
} catch (IllegalAccessException e) {
log.error("Class couldn't be instantiated due to IllegalAccessException: " + className, e);
throw new CopycatRuntimeException(e);
}
}
/**
* Instantiate the class, which is a subclass of the specified type, and start it.
* @param className fully qualified name of the class to instantiate
* @param superType required supertype of the class
* @param <T>
* @return a new instance of the class instantiated with the default constructor
*/
public static <T extends Configurable> T instantiateConfigurable(
String className, Class<T> superType, Properties props) {
T result = instantiate(className, superType);
result.configure(props);
return result;
}
}