You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by ki...@apache.org on 2019/12/10 20:35:46 UTC
[incubator-pinot] 01/01: Replacing usage of Class.forName with
PluginManager.createInstance
This is an automated email from the ASF dual-hosted git repository.
kishoreg pushed a commit to branch use-plugin-manager
in repository https://gitbox.apache.org/repos/asf/incubator-pinot.git
commit ed3687f8bf26e2531c29f52c0b367626bfd69a0f
Author: kishoreg <g....@gmail.com>
AuthorDate: Tue Dec 10 12:34:33 2019 -0800
Replacing usage of Class.forName with PluginManager.createInstance
---
.../common/SegmentGenerationTaskRunner.java | 8 +++--
.../apache/pinot/filesystem/PinotFSFactory.java | 5 +--
.../org/apache/pinot/common/data/SchemaTest.java | 5 ++-
.../ThriftRecordReader.java | 2 +-
.../pinot/spi/crypt/PinotCrypterFactory.java | 6 ++--
.../java/org/apache/pinot/spi/data/Schema.java | 17 +++++++++
.../spi/data/readers/RecordReaderFactory.java | 40 +++++++++-------------
.../org/apache/pinot/spi/plugin/PluginManager.java | 24 ++++++++-----
.../spi/stream/StreamConsumerFactoryProvider.java | 3 +-
.../pinot/spi/stream/StreamDataProvider.java | 7 ++--
.../pinot/spi/stream/StreamDecoderProvider.java | 3 +-
11 files changed, 72 insertions(+), 48 deletions(-)
diff --git a/pinot-batch-ingestion/pinot-standalone/src/main/java/org/apache/pinot/ingestion/common/SegmentGenerationTaskRunner.java b/pinot-batch-ingestion/pinot-standalone/src/main/java/org/apache/pinot/ingestion/common/SegmentGenerationTaskRunner.java
index 5d5b871..49e2f4e 100644
--- a/pinot-batch-ingestion/pinot-standalone/src/main/java/org/apache/pinot/ingestion/common/SegmentGenerationTaskRunner.java
+++ b/pinot-batch-ingestion/pinot-standalone/src/main/java/org/apache/pinot/ingestion/common/SegmentGenerationTaskRunner.java
@@ -35,6 +35,7 @@ import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.data.TimeFieldSpec;
import org.apache.pinot.spi.data.readers.RecordReader;
import org.apache.pinot.spi.data.readers.RecordReaderConfig;
+import org.apache.pinot.spi.plugin.PluginManager;
import org.apache.pinot.spi.utils.JsonUtils;
@@ -72,13 +73,14 @@ public class SegmentGenerationTaskRunner {
configs = new HashMap<>();
}
JsonNode jsonNode = new ObjectMapper().valueToTree(configs);
- JsonUtils.jsonNodeToObject(jsonNode, Class.forName(readerConfigClassName));
- recordReaderConfig = (RecordReaderConfig) Class.forName(readerConfigClassName).newInstance();
+ Class<?> clazz = PluginManager.get().loadClass(readerConfigClassName);
+ recordReaderConfig = (RecordReaderConfig) JsonUtils.jsonNodeToObject(jsonNode, clazz);
}
//init record reader
String readerClassName = _taskSpec.getRecordReaderSpec().getClassName();
- RecordReader recordReader = (RecordReader) Class.forName(readerClassName).newInstance();
+ RecordReader recordReader = PluginManager.get().createInstance(readerClassName);
+
recordReader.init(new File(_taskSpec.getInputFilePath()), schema, recordReaderConfig);
//init segmentName Generator
diff --git a/pinot-common/src/main/java/org/apache/pinot/filesystem/PinotFSFactory.java b/pinot-common/src/main/java/org/apache/pinot/filesystem/PinotFSFactory.java
index 935707f..bbf2f43 100644
--- a/pinot-common/src/main/java/org/apache/pinot/filesystem/PinotFSFactory.java
+++ b/pinot-common/src/main/java/org/apache/pinot/filesystem/PinotFSFactory.java
@@ -24,6 +24,7 @@ import java.util.Iterator;
import java.util.Map;
import org.apache.commons.configuration.Configuration;
import org.apache.pinot.spi.filesystem.PinotFS;
+import org.apache.pinot.spi.plugin.PluginManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -46,10 +47,10 @@ public class PinotFSFactory {
public static void register(String scheme, String fsClassName, Configuration configuration) {
try {
LOGGER.info("Initializing PinotFS for scheme {}, classname {}", scheme, fsClassName);
- PinotFS pinotFS = (PinotFS) Class.forName(fsClassName).newInstance();
+ PinotFS pinotFS = PluginManager.get().createInstance(fsClassName);
pinotFS.init(configuration);
_fileSystemMap.put(scheme, pinotFS);
- } catch (InstantiationException | IllegalAccessException | ClassNotFoundException e) {
+ } catch (Exception e) {
LOGGER.error("Could not instantiate file system for class {} with scheme {}", fsClassName, scheme, e);
throw new RuntimeException(e);
}
diff --git a/pinot-common/src/test/java/org/apache/pinot/common/data/SchemaTest.java b/pinot-common/src/test/java/org/apache/pinot/common/data/SchemaTest.java
index cbc593a..262707d 100644
--- a/pinot-common/src/test/java/org/apache/pinot/common/data/SchemaTest.java
+++ b/pinot-common/src/test/java/org/apache/pinot/common/data/SchemaTest.java
@@ -79,7 +79,10 @@ public class SchemaTest {
.addMetric("derivedMetric", FieldSpec.DataType.STRING, 10, MetricFieldSpec.DerivedMetricType.HLL)
.addMetric("derivedMetricWithDefault", FieldSpec.DataType.STRING, 10, MetricFieldSpec.DerivedMetricType.HLL,
defaultString).addTime("time", TimeUnit.DAYS, FieldSpec.DataType.LONG)
- .addDateTime("dateTime", FieldSpec.DataType.LONG, "1:HOURS:EPOCH", "1:HOURS").build();
+ .addDateTime("launch_date", FieldSpec.DataType.LONG, "1:HOURS:EPOCH", "1:HOURS")
+ .addDateTime("launch_date", FieldSpec.DataType.LONG, "1:HOURS:EPOCH", "1:HOURS")
+ .addDateTime("due_date", FieldSpec.DataType.LONG, "1:HOURS:EPOCH", "1:HOURS")
+ .build();
DimensionFieldSpec dimensionFieldSpec = schema.getDimensionSpec("svDimension");
Assert.assertNotNull(dimensionFieldSpec);
diff --git a/pinot-record-readers/pinot-thrift/src/main/java/org.apache.pinot.thrift.data.readers/ThriftRecordReader.java b/pinot-record-readers/pinot-thrift/src/main/java/org.apache.pinot.thrift.data.readers/ThriftRecordReader.java
index fd16e72..b97d7fe 100644
--- a/pinot-record-readers/pinot-thrift/src/main/java/org.apache.pinot.thrift.data.readers/ThriftRecordReader.java
+++ b/pinot-record-readers/pinot-thrift/src/main/java/org.apache.pinot.thrift.data.readers/ThriftRecordReader.java
@@ -84,7 +84,7 @@ public class ThriftRecordReader implements RecordReader {
_fieldSpecs = RecordReaderUtils.extractFieldSpecs(schema);
TBase tObject;
try {
- _thriftClass = Class.forName(recordReaderConfig.getThriftClass());
+ _thriftClass = this.getClass().getClassLoader().loadClass(recordReaderConfig.getThriftClass());
tObject = (TBase) _thriftClass.newInstance();
} catch (Exception e) {
throw new RuntimeException(e);
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/crypt/PinotCrypterFactory.java b/pinot-spi/src/main/java/org/apache/pinot/spi/crypt/PinotCrypterFactory.java
index 5f41068..f4a4d69 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/crypt/PinotCrypterFactory.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/crypt/PinotCrypterFactory.java
@@ -22,6 +22,7 @@ import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import org.apache.commons.configuration.Configuration;
+import org.apache.pinot.spi.plugin.PluginManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -60,12 +61,11 @@ public class PinotCrypterFactory {
LOGGER.info("Got crypter class name {}, full crypter path {}, starting to initialize", key, className);
try {
- PinotCrypter pinotCrypter = (PinotCrypter) Class.forName(className).newInstance();
+ PinotCrypter pinotCrypter = PluginManager.get().createInstance(className);
pinotCrypter.init(config.subset(key));
-
LOGGER.info("Initializing PinotCrypter for scheme {}, classname {}", key, className);
_crypterMap.put(key.toLowerCase(), pinotCrypter);
- } catch (InstantiationException | IllegalAccessException | ClassNotFoundException e) {
+ } catch (Exception e) {
LOGGER.error("Could not instantiate crypter for class {}", className, e);
throw new RuntimeException(e);
}
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/data/Schema.java b/pinot-spi/src/main/java/org/apache/pinot/spi/data/Schema.java
index f0857c0..cf27c81 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/data/Schema.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/data/Schema.java
@@ -626,4 +626,21 @@ public final class Schema {
result = EqualityUtils.hashCodeOf(result, _dateTimeFieldSpecs);
return result;
}
+
+ public static void main(String[] args) {
+ String defaultString = "";
+ Schema schema = new Schema.SchemaBuilder().addSingleValueDimension("svDimension", FieldSpec.DataType.INT)
+ .addSingleValueDimension("svDimensionWithDefault", FieldSpec.DataType.INT, 10)
+ .addMultiValueDimension("mvDimension", FieldSpec.DataType.STRING)
+ .addMultiValueDimension("mvDimensionWithDefault", FieldSpec.DataType.STRING, defaultString)
+ .addMetric("metric", FieldSpec.DataType.INT).addMetric("metricWithDefault", FieldSpec.DataType.INT, 5)
+ .addMetric("derivedMetric", FieldSpec.DataType.STRING, 10, MetricFieldSpec.DerivedMetricType.HLL)
+ .addMetric("derivedMetricWithDefault", FieldSpec.DataType.STRING, 10, MetricFieldSpec.DerivedMetricType.HLL,
+ defaultString).addTime("time", TimeUnit.DAYS, FieldSpec.DataType.LONG)
+ .addDateTime("payment_date", FieldSpec.DataType.LONG, "1:HOURS:EPOCH", "1:HOURS")
+ .addDateTime("launch_date", FieldSpec.DataType.LONG, "1:HOURS:EPOCH", "1:HOURS")
+ .addDateTime("due_date", FieldSpec.DataType.LONG, "1:HOURS:EPOCH", "1:HOURS")
+ .build();
+ System.out.println("schema.toPrettyJsonString() = " + schema.toPrettyJsonString());
+ }
}
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/data/readers/RecordReaderFactory.java b/pinot-spi/src/main/java/org/apache/pinot/spi/data/readers/RecordReaderFactory.java
index 0a0ff1d..f5497d8 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/data/readers/RecordReaderFactory.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/data/readers/RecordReaderFactory.java
@@ -23,6 +23,7 @@ import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.plugin.PluginManager;
import org.apache.pinot.spi.utils.JsonUtils;
@@ -66,7 +67,8 @@ public class RecordReaderFactory {
* @throws IOException
* @throws ClassNotFoundException
*/
- public static RecordReaderConfig getRecordReaderConfigByClassName(String recordReaderConfigClassName, String readerConfigFile)
+ public static RecordReaderConfig getRecordReaderConfigByClassName(String recordReaderConfigClassName,
+ String readerConfigFile)
throws IOException, ClassNotFoundException {
return getRecordReaderConfigByClassName(recordReaderConfigClassName, new File(readerConfigFile));
}
@@ -80,9 +82,10 @@ public class RecordReaderFactory {
* @throws IOException
* @throws ClassNotFoundException
*/
- public static RecordReaderConfig getRecordReaderConfigByClassName(String recordReaderConfigClassName, File readerConfigFile)
+ public static RecordReaderConfig getRecordReaderConfigByClassName(String recordReaderConfigClassName,
+ File readerConfigFile)
throws IOException, ClassNotFoundException {
- Class recordReaderConfigClass = Class.forName(recordReaderConfigClassName);
+ Class recordReaderConfigClass = PluginManager.get().loadClass(recordReaderConfigClassName);
RecordReaderConfig recordReaderConfig =
(RecordReaderConfig) JsonUtils.fileToObject(readerConfigFile, recordReaderConfigClass);
return recordReaderConfig;
@@ -94,14 +97,14 @@ public class RecordReaderFactory {
* @param fileFormat
* @param readerConfigFile
* @return a RecordReaderConfig instance
- * @throws IOException
- * @throws ClassNotFoundException
+ * @throws Exception
*/
public static RecordReaderConfig getRecordReaderConfig(FileFormat fileFormat, String readerConfigFile)
- throws IOException, ClassNotFoundException {
+ throws Exception {
String fileFormatKey = fileFormat.name().toUpperCase();
if (DEFAULT_RECORD_READER_CONFIG_CLASS_MAP.containsKey(fileFormatKey)) {
- return getRecordReaderConfigByClassName(DEFAULT_RECORD_READER_CONFIG_CLASS_MAP.get(fileFormatKey), readerConfigFile);
+ return getRecordReaderConfigByClassName(DEFAULT_RECORD_READER_CONFIG_CLASS_MAP.get(fileFormatKey),
+ readerConfigFile);
}
throw new UnsupportedOperationException("No supported RecordReader found for file format - '" + fileFormat + "'");
}
@@ -114,15 +117,12 @@ public class RecordReaderFactory {
* @param schema
* @param recordReaderConfig
* @return an initialized RecordReader instance.
- * @throws IOException
- * @throws ClassNotFoundException
- * @throws IllegalAccessException
- * @throws InstantiationException
+ * @throws Exception
*/
public static RecordReader getRecordReaderByClass(String recordReaderClassName, File dataFile, Schema schema,
RecordReaderConfig recordReaderConfig)
- throws IOException, ClassNotFoundException, IllegalAccessException, InstantiationException {
- RecordReader recordReader = (RecordReader) Class.forName(recordReaderClassName).newInstance();
+ throws Exception {
+ RecordReader recordReader = PluginManager.get().createInstance(recordReaderClassName);
recordReader.init(dataFile, schema, recordReaderConfig);
return recordReader;
}
@@ -135,14 +135,11 @@ public class RecordReaderFactory {
* @param schema
* @param recordReaderConfig
* @return an initialized RecordReader instance.
- * @throws ClassNotFoundException
- * @throws IOException
- * @throws InstantiationException
- * @throws IllegalAccessException
+ * @throws Exception Any exception while initializing the RecordReader
*/
public static RecordReader getRecordReader(FileFormat fileFormat, File dataFile, Schema schema,
RecordReaderConfig recordReaderConfig)
- throws ClassNotFoundException, IOException, InstantiationException, IllegalAccessException {
+ throws Exception {
return getRecordReader(fileFormat.name(), dataFile, schema, recordReaderConfig);
}
@@ -154,14 +151,11 @@ public class RecordReaderFactory {
* @param schema
* @param recordReaderConfig
* @return an initialized RecordReader instance.
- * @throws ClassNotFoundException
- * @throws InstantiationException
- * @throws IllegalAccessException
- * @throws IOException
+ * @throws Exception Any exception while initializing the RecordReader
*/
public static RecordReader getRecordReader(String fileFormatStr, File dataFile, Schema schema,
RecordReaderConfig recordReaderConfig)
- throws ClassNotFoundException, InstantiationException, IllegalAccessException, IOException {
+ throws Exception {
String fileFormatKey = fileFormatStr.toUpperCase();
if (DEFAULT_RECORD_READER_CLASS_MAP.containsKey(fileFormatKey)) {
return getRecordReaderByClass(DEFAULT_RECORD_READER_CLASS_MAP.get(fileFormatKey), dataFile, schema,
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/plugin/PluginManager.java b/pinot-spi/src/main/java/org/apache/pinot/spi/plugin/PluginManager.java
index d6b5458..a28a916 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/plugin/PluginManager.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/plugin/PluginManager.java
@@ -20,6 +20,7 @@ package org.apache.pinot.spi.plugin;
import java.io.File;
import java.lang.reflect.Constructor;
+import java.lang.reflect.InvocationTargetException;
import java.net.MalformedURLException;
import java.net.URL;
import java.util.ArrayList;
@@ -29,6 +30,7 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import org.apache.commons.io.FileUtils;
+import org.apache.commons.lang3.exception.ExceptionUtils;
public class PluginManager {
@@ -107,7 +109,6 @@ public class PluginManager {
* <li>pluginName:com.x.y.foo</li> loads the class in plugin specific classloader
* @param className
* @return
- * @throws ClassNotFoundException
*/
public <T> T createInstance(String className)
throws Exception {
@@ -120,7 +121,6 @@ public class PluginManager {
* <li>pluginName:com.x.y.foo</li> loads the class in plugin specific classloader
* @param className
* @return
- * @throws ClassNotFoundException
*/
public <T> T createInstance(String className, Class[] argTypes, Object[] argValues)
throws Exception {
@@ -156,14 +156,20 @@ public class PluginManager {
* @param <T>
* @return
*/
- public <T> T createInstance(String pluginName, String className, Class[] argTypes, Object[] argValues)
- throws Exception {
+ public <T> T createInstance(String pluginName, String className, Class[] argTypes, Object[] argValues) throws Exception{
PluginClassLoader pluginClassLoader = PLUGIN_MANAGER._registry.get(new Plugin(pluginName));
- Class<T> loadedClass = (Class<T>) pluginClassLoader.loadClass(className, true);
- Constructor<?> constructor = loadedClass.getConstructor(argTypes);
- if (constructor != null) {
- Object instance = constructor.newInstance(argValues);
- return (T) instance;
+ try {
+ Class<T> loadedClass = (Class<T>) pluginClassLoader.loadClass(className, true);
+ Constructor<?> constructor = null;
+ constructor = loadedClass.getConstructor(argTypes);
+ if (constructor != null) {
+ Object instance = null;
+ instance = constructor.newInstance(argValues);
+
+ return (T) instance;
+ }
+ } catch (NoSuchMethodException | InstantiationException | IllegalAccessException | InvocationTargetException | ClassNotFoundException e) {
+ throw e;
}
return null;
}
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamConsumerFactoryProvider.java b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamConsumerFactoryProvider.java
index 824c439..5cfe525 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamConsumerFactoryProvider.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamConsumerFactoryProvider.java
@@ -19,6 +19,7 @@
package org.apache.pinot.spi.stream;
import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.apache.pinot.spi.plugin.PluginManager;
/**
@@ -34,7 +35,7 @@ public abstract class StreamConsumerFactoryProvider {
public static StreamConsumerFactory create(StreamConfig streamConfig) {
StreamConsumerFactory factory = null;
try {
- factory = (StreamConsumerFactory) Class.forName(streamConfig.getConsumerFactoryClassName()).newInstance();
+ factory = PluginManager.get().createInstance(streamConfig.getConsumerFactoryClassName());
} catch (Exception e) {
ExceptionUtils.rethrow(e);
}
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamDataProvider.java b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamDataProvider.java
index 3889da4..5a9506a 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamDataProvider.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamDataProvider.java
@@ -19,6 +19,7 @@
package org.apache.pinot.spi.stream;
import java.util.Properties;
+import org.apache.pinot.spi.plugin.PluginManager;
/**
@@ -30,16 +31,14 @@ import java.util.Properties;
public class StreamDataProvider {
public static StreamDataServerStartable getServerDataStartable(String clazz, Properties props)
throws Exception {
- final StreamDataServerStartable streamDataServerStartable =
- (StreamDataServerStartable) Class.forName(clazz).newInstance();
+ final StreamDataServerStartable streamDataServerStartable = PluginManager.get().createInstance(clazz);
streamDataServerStartable.init(props);
return streamDataServerStartable;
}
public static StreamDataProducer getStreamDataProducer(String clazz, Properties props)
throws Exception {
-
- final StreamDataProducer streamDataProducer = (StreamDataProducer) Class.forName(clazz).newInstance();
+ final StreamDataProducer streamDataProducer = PluginManager.get().createInstance(clazz);;
streamDataProducer.init(props);
return streamDataProducer;
}
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamDecoderProvider.java b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamDecoderProvider.java
index d6b8114..e4cac8a 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamDecoderProvider.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamDecoderProvider.java
@@ -21,6 +21,7 @@ package org.apache.pinot.spi.stream;
import java.util.Map;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.plugin.PluginManager;
/**
@@ -39,7 +40,7 @@ public abstract class StreamDecoderProvider {
String decoderClass = streamConfig.getDecoderClass();
Map<String, String> decoderProperties = streamConfig.getDecoderProperties();
try {
- decoder = (StreamMessageDecoder) Class.forName(decoderClass).newInstance();
+ decoder = PluginManager.get().createInstance(decoderClass);
decoder.init(decoderProperties, schema, streamConfig.getTopicName());
} catch (Exception e) {
ExceptionUtils.rethrow(e);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org