You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@eagle.apache.org by ha...@apache.org on 2016/09/08 12:24:40 UTC
[2/3] incubator-eagle git commit: [EAGLE-533] Fix storage
configuration and remove AppJUnitRunner
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/c752ed86/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/service/impl/ApplicationProviderConfigLoader.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/service/impl/ApplicationProviderConfigLoader.java b/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/service/impl/ApplicationProviderConfigLoader.java
index 3b2bca9..2f613c1 100644
--- a/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/service/impl/ApplicationProviderConfigLoader.java
+++ b/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/service/impl/ApplicationProviderConfigLoader.java
@@ -16,25 +16,26 @@
*/
package org.apache.eagle.app.service.impl;
-import com.google.common.base.Preconditions;
-import com.typesafe.config.Config;
import org.apache.eagle.app.config.ApplicationProviderConfig;
import org.apache.eagle.app.config.ApplicationProvidersConfig;
import org.apache.eagle.app.service.ApplicationProviderLoader;
import org.apache.eagle.app.spi.ApplicationProvider;
+import com.google.common.base.Preconditions;
+import com.typesafe.config.Config;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import javax.xml.bind.JAXBContext;
-import javax.xml.bind.Unmarshaller;
import java.io.IOException;
import java.io.InputStream;
import java.util.List;
+import javax.xml.bind.JAXBContext;
+import javax.xml.bind.Unmarshaller;
public class ApplicationProviderConfigLoader extends ApplicationProviderLoader {
- public final static String DEFAULT_APPLICATIONS_CONFIG_FILE = "providers.xml";
- private final static String APPLICATIONS_CONFIG_PROPS_KEY = "application.provider.config";
- private final static Logger LOG = LoggerFactory.getLogger(ApplicationProviderConfigLoader.class);
+ public static final String DEFAULT_APPLICATIONS_CONFIG_FILE = "providers.xml";
+ private static final String APPLICATIONS_CONFIG_PROPS_KEY = "application.provider.config";
+ private static final Logger LOG = LoggerFactory.getLogger(ApplicationProviderConfigLoader.class);
+
public ApplicationProviderConfigLoader(Config config) {
super(config);
}
@@ -43,26 +44,27 @@ public class ApplicationProviderConfigLoader extends ApplicationProviderLoader {
public void load() {
List<ApplicationProviderConfig> applicationProviderConfigs = loadProvidersFromProvidersConf();
int totalCount = applicationProviderConfigs.size();
- int loadedCount = 0,failedCount = 0;
- for(ApplicationProviderConfig providerConfig: applicationProviderConfigs){
+ int loadedCount = 0;
+ int failedCount = 0;
+ for (ApplicationProviderConfig providerConfig : applicationProviderConfigs) {
try {
initializeProvider(providerConfig);
- loadedCount ++;
- }catch (Throwable ex){
- LOG.warn("Failed to initialized {}, ignored",providerConfig,ex);
- failedCount ++;
+ loadedCount++;
+ } catch (Throwable ex) {
+ LOG.warn("Failed to initialized {}, ignored", providerConfig, ex);
+ failedCount++;
}
}
- LOG.info("Loaded {} app providers (total: {}, failed: {})",loadedCount,totalCount,failedCount);
+ LOG.info("Loaded {} app providers (total: {}, failed: {})", loadedCount, totalCount, failedCount);
}
- public static boolean appProviderConfExists(String applicationConfFile){
+ public static boolean appProviderConfExists(String applicationConfFile) {
InputStream is = ApplicationProviderConfigLoader.class.getResourceAsStream(applicationConfFile);
- if(is == null){
- is = ApplicationProviderConfigLoader.class.getResourceAsStream("/"+applicationConfFile);
+ if (is == null) {
+ is = ApplicationProviderConfigLoader.class.getResourceAsStream("/" + applicationConfFile);
}
- if(is != null){
+ if (is != null) {
try {
return true;
} finally {
@@ -78,50 +80,56 @@ public class ApplicationProviderConfigLoader extends ApplicationProviderLoader {
}
private void initializeProvider(ApplicationProviderConfig providerConfig) throws ClassNotFoundException, IllegalAccessException, InstantiationException {
- LOG.info("Loading application provider {} from {}",providerConfig.getClassName(),providerConfig.getJarPath());
+ LOG.info("Loading application provider {} from {}", providerConfig.getClassName(), providerConfig.getJarPath());
String providerClassName = providerConfig.getClassName();
- if(providerClassName == null) throw new RuntimeException("provider.classname is null: "+providerConfig);
- if(providerConfig.getJarPath() == null) throw new RuntimeException("provider.jarpath is null: "+providerConfig);
+ if (providerClassName == null) {
+ throw new RuntimeException("provider.classname is null: " + providerConfig);
+ }
+ if (providerConfig.getJarPath() == null) {
+ throw new RuntimeException("provider.jarpath is null: " + providerConfig);
+ }
Class<?> providerClass = Class.forName(providerClassName);
- if(!ApplicationProvider.class.isAssignableFrom(providerClass)){
- throw new RuntimeException("providerClassName is not implementation of "+ApplicationProvider.class.getCanonicalName());
+ if (!ApplicationProvider.class.isAssignableFrom(providerClass)) {
+ throw new RuntimeException("providerClassName is not implementation of " + ApplicationProvider.class.getCanonicalName());
}
ApplicationProvider provider = (ApplicationProvider) providerClass.newInstance();
- provider.prepare(providerConfig,this.getConfig());
- Preconditions.checkNotNull(provider.getApplicationDesc(),"appDesc is null");
- Preconditions.checkNotNull(provider.getApplicationDesc().getType(),"type is null");
+ provider.prepare(providerConfig, this.getConfig());
+ Preconditions.checkNotNull(provider.getApplicationDesc(), "appDesc is null");
+ Preconditions.checkNotNull(provider.getApplicationDesc().getType(), "type is null");
registerProvider(provider);
}
private List<ApplicationProviderConfig> loadProvidersFromProvidersConf() {
String providerConfigFile = DEFAULT_APPLICATIONS_CONFIG_FILE;
- if(getConfig().hasPath(APPLICATIONS_CONFIG_PROPS_KEY)){
+ if (getConfig().hasPath(APPLICATIONS_CONFIG_PROPS_KEY)) {
providerConfigFile = getConfig().getString(APPLICATIONS_CONFIG_PROPS_KEY);
- LOG.info("Set {} = {}",APPLICATIONS_CONFIG_PROPS_KEY,providerConfigFile);
+ LOG.info("Set {} = {}", APPLICATIONS_CONFIG_PROPS_KEY, providerConfigFile);
}
InputStream is = null;
try {
JAXBContext jc = JAXBContext.newInstance(ApplicationProvidersConfig.class);
- Unmarshaller unmarshaller = jc.createUnmarshaller();
+ final Unmarshaller unmarshaller = jc.createUnmarshaller();
is = ApplicationProviderConfigLoader.class.getResourceAsStream(providerConfigFile);
- if(is == null){
- is = ApplicationProviderConfigLoader.class.getResourceAsStream("/"+providerConfigFile);
+ if (is == null) {
+ is = ApplicationProviderConfigLoader.class.getResourceAsStream("/" + providerConfigFile);
}
- if(is == null){
- LOG.error("Application provider configuration {} is not found",providerConfigFile);
+ if (is == null) {
+ LOG.error("Application provider configuration {} is not found", providerConfigFile);
}
- Preconditions.checkNotNull(is,providerConfigFile+" is not found");
+ Preconditions.checkNotNull(is, providerConfigFile + " is not found");
return ((ApplicationProvidersConfig) unmarshaller.unmarshal(is)).getProviders();
- }catch (Exception ex){
- LOG.error("Failed to load application provider configuration: {}",providerConfigFile,ex);
- throw new RuntimeException("Failed to load application provider configuration: "+providerConfigFile,ex);
+ } catch (Exception ex) {
+ LOG.error("Failed to load application provider configuration: {}", providerConfigFile, ex);
+ throw new RuntimeException("Failed to load application provider configuration: " + providerConfigFile, ex);
} finally {
- if(is != null) try {
- is.close();
- } catch (IOException e) {
- LOG.error(e.getMessage(),e);
+ if (is != null) {
+ try {
+ is.close();
+ } catch (IOException e) {
+ LOG.error(e.getMessage(), e);
+ }
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/c752ed86/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/service/impl/ApplicationProviderSPILoader.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/service/impl/ApplicationProviderSPILoader.java b/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/service/impl/ApplicationProviderSPILoader.java
index c7bc4f6..8b73865 100644
--- a/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/service/impl/ApplicationProviderSPILoader.java
+++ b/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/service/impl/ApplicationProviderSPILoader.java
@@ -16,11 +16,11 @@
*/
package org.apache.eagle.app.service.impl;
-import com.typesafe.config.Config;
import org.apache.eagle.app.config.ApplicationProviderConfig;
import org.apache.eagle.app.service.ApplicationProviderLoader;
import org.apache.eagle.app.spi.ApplicationProvider;
import org.apache.eagle.app.utils.DynamicJarPathFinder;
+import com.typesafe.config.Config;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -30,24 +30,24 @@ import java.net.URLClassLoader;
import java.util.ServiceLoader;
import java.util.function.Function;
-public class ApplicationProviderSPILoader extends ApplicationProviderLoader{
+public class ApplicationProviderSPILoader extends ApplicationProviderLoader {
private final String appProviderExtDir;
- private final static Logger LOG = LoggerFactory.getLogger(ApplicationProviderSPILoader.class);
- private final static String APPLICATIONS_DIR_PROPS_KEY = "application.provider.dir";
+ private static final Logger LOG = LoggerFactory.getLogger(ApplicationProviderSPILoader.class);
+ private static final String APPLICATIONS_DIR_PROPS_KEY = "application.provider.dir";
public ApplicationProviderSPILoader(Config config) {
super(config);
- if(config.hasPath(APPLICATIONS_DIR_PROPS_KEY)) {
+ if (config.hasPath(APPLICATIONS_DIR_PROPS_KEY)) {
this.appProviderExtDir = config.getString(APPLICATIONS_DIR_PROPS_KEY);
- LOG.warn("Using {}: {}",APPLICATIONS_DIR_PROPS_KEY,this.appProviderExtDir);
- }else{
+ LOG.warn("Using {}: {}", APPLICATIONS_DIR_PROPS_KEY, this.appProviderExtDir);
+ } else {
this.appProviderExtDir = null;
}
}
@Override
public void load() {
- if(appProviderExtDir != null) {
+ if (appProviderExtDir != null) {
LOG.warn("Loading application providers from class loader of jars in {}", appProviderExtDir);
File loc = new File(appProviderExtDir);
File[] jarFiles = loc.listFiles(file -> file.getPath().toLowerCase().endsWith(".jar"));
@@ -56,21 +56,21 @@ public class ApplicationProviderSPILoader extends ApplicationProviderLoader{
try {
URL jarFileUrl = jarFile.toURI().toURL();
LOG.debug("Loading ApplicationProvider from jar: {}", jarFileUrl.toString());
- URLClassLoader jarFileClassLoader = new URLClassLoader(new URL[]{jarFileUrl});
+ URLClassLoader jarFileClassLoader = new URLClassLoader(new URL[] {jarFileUrl});
loadProviderFromClassLoader(jarFileClassLoader, (applicationProviderConfig) -> jarFileUrl.getPath());
} catch (Exception e) {
- LOG.warn("Failed to load application provider from jar {}", jarFile,e);
+ LOG.warn("Failed to load application provider from jar {}", jarFile, e);
}
}
}
} else {
LOG.warn("Loading application providers from context class loader");
ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
- loadProviderFromClassLoader(classLoader,(applicationProvider) -> DynamicJarPathFinder.findPath(applicationProvider.getClass()));
+ loadProviderFromClassLoader(classLoader, (applicationProvider) -> DynamicJarPathFinder.findPath(applicationProvider.getClass()));
}
}
- private void loadProviderFromClassLoader(ClassLoader jarFileClassLoader, Function<ApplicationProvider,String> jarFileSupplier){
+ private void loadProviderFromClassLoader(ClassLoader jarFileClassLoader, Function<ApplicationProvider, String> jarFileSupplier) {
ServiceLoader<ApplicationProvider> serviceLoader = ServiceLoader.load(ApplicationProvider.class);
try {
for (ApplicationProvider applicationProvider : serviceLoader) {
@@ -80,8 +80,8 @@ public class ApplicationProviderSPILoader extends ApplicationProviderLoader{
applicationProvider.prepare(providerConfig, getConfig());
registerProvider(applicationProvider);
}
- }catch (Throwable ex){
- LOG.warn("Failed to register application provider",ex);
+ } catch (Throwable ex) {
+ LOG.warn("Failed to register application provider", ex);
throw new IllegalStateException(ex);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/c752ed86/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/service/impl/ApplicationProviderServiceImpl.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/service/impl/ApplicationProviderServiceImpl.java b/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/service/impl/ApplicationProviderServiceImpl.java
index ec2ad6c..3cb31f9 100644
--- a/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/service/impl/ApplicationProviderServiceImpl.java
+++ b/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/service/impl/ApplicationProviderServiceImpl.java
@@ -16,65 +16,64 @@
*/
package org.apache.eagle.app.service.impl;
-import com.google.inject.Inject;
-import com.google.inject.Singleton;
-import com.typesafe.config.Config;
import org.apache.eagle.app.service.ApplicationProviderLoader;
import org.apache.eagle.app.service.ApplicationProviderService;
import org.apache.eagle.app.spi.ApplicationProvider;
import org.apache.eagle.metadata.model.ApplicationDesc;
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+import com.typesafe.config.Config;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.*;
+import java.util.Collection;
import java.util.stream.Collectors;
/**
* Support to load application provider from application.provider.config = "providers.xml" configuration file
* or application.provider.dir = "lib/apps" with SPI Class loader
- *
- * TODO: hot-manage application provider loading
+ * <p>TODO: hot-manage application provider loading</p>
*/
@Singleton
public class ApplicationProviderServiceImpl implements ApplicationProviderService {
private final Config config;
- private final static Logger LOG = LoggerFactory.getLogger(ApplicationProviderServiceImpl.class);
+ private static final Logger LOG = LoggerFactory.getLogger(ApplicationProviderServiceImpl.class);
private final ApplicationProviderLoader appProviderLoader;
- public final static String APP_PROVIDER_LOADER_CLASS_KEY = "application.provider.loader";
+ public static final String APP_PROVIDER_LOADER_CLASS_KEY = "application.provider.loader";
@Inject
- public ApplicationProviderServiceImpl(Config config){
- LOG.warn("Initializing {}",this.getClass().getCanonicalName());
+ public ApplicationProviderServiceImpl(Config config) {
+ LOG.warn("Initializing {}", this.getClass().getCanonicalName());
this.config = config;
- String appProviderLoaderClass = this.config.hasPath(APP_PROVIDER_LOADER_CLASS_KEY)?
- this.config.getString(APP_PROVIDER_LOADER_CLASS_KEY):ApplicationProviderLoader.getDefaultAppProviderLoader();
- LOG.warn("Initializing {} = {}",APP_PROVIDER_LOADER_CLASS_KEY,appProviderLoaderClass);
+ String appProviderLoaderClass = this.config.hasPath(APP_PROVIDER_LOADER_CLASS_KEY)
+ ? this.config.getString(APP_PROVIDER_LOADER_CLASS_KEY) : ApplicationProviderLoader.getDefaultAppProviderLoader();
+ LOG.warn("Initializing {} = {}", APP_PROVIDER_LOADER_CLASS_KEY, appProviderLoaderClass);
appProviderLoader = initializeAppProviderLoader(appProviderLoaderClass);
- LOG.warn("Initialized {}",appProviderLoader);
+ LOG.warn("Initialized {}", appProviderLoader);
reload();
}
- private ApplicationProviderLoader initializeAppProviderLoader(String appProviderLoaderClass){
+ private ApplicationProviderLoader initializeAppProviderLoader(String appProviderLoaderClass) {
try {
return (ApplicationProviderLoader) Class.forName(appProviderLoaderClass).getConstructor(Config.class).newInstance(this.config);
} catch (Throwable e) {
- LOG.error("Failed to initialize ApplicationProviderLoader: "+appProviderLoaderClass,e);
- throw new IllegalStateException("Failed to initialize ApplicationProviderLoader: "+appProviderLoaderClass,e);
+ LOG.error("Failed to initialize ApplicationProviderLoader: " + appProviderLoaderClass, e);
+ throw new IllegalStateException("Failed to initialize ApplicationProviderLoader: " + appProviderLoaderClass, e);
}
}
- public synchronized void reload(){
+ public synchronized void reload() {
appProviderLoader.reset();
LOG.warn("Loading application providers ...");
appProviderLoader.load();
- LOG.warn("Loaded {} application providers",appProviderLoader.getProviders().size());
+ LOG.warn("Loaded {} application providers", appProviderLoader.getProviders().size());
}
- public Collection<ApplicationProvider> getProviders(){
+ public Collection<ApplicationProvider> getProviders() {
return appProviderLoader.getProviders();
}
- public Collection<ApplicationDesc> getApplicationDescs(){
+ public Collection<ApplicationDesc> getApplicationDescs() {
return getProviders().stream().map(ApplicationProvider::getApplicationDesc).collect(Collectors.toList());
}
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/c752ed86/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/sink/DefaultStreamSinkConfig.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/sink/DefaultStreamSinkConfig.java b/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/sink/DefaultStreamSinkConfig.java
index 4a20e2b..b3d984e 100644
--- a/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/sink/DefaultStreamSinkConfig.java
+++ b/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/sink/DefaultStreamSinkConfig.java
@@ -20,9 +20,9 @@ import org.apache.eagle.metadata.model.StreamSinkConfig;
public class DefaultStreamSinkConfig implements StreamSinkConfig {
private final Class<?> streamPersistClass;
- private final static String NONE_STORAGE_TYPE = "NONE";
+ private static final String NONE_STORAGE_TYPE = "NONE";
- public DefaultStreamSinkConfig(Class<?> streamPersistClass){
+ public DefaultStreamSinkConfig(Class<?> streamPersistClass) {
this.streamPersistClass = streamPersistClass;
}
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/c752ed86/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/sink/FlattenEventMapper.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/sink/FlattenEventMapper.java b/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/sink/FlattenEventMapper.java
index fe3562b..3649d72 100644
--- a/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/sink/FlattenEventMapper.java
+++ b/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/sink/FlattenEventMapper.java
@@ -16,39 +16,39 @@
*/
package org.apache.eagle.app.sink;
-import backtype.storm.tuple.Tuple;
import org.apache.eagle.alert.engine.model.StreamEvent;
+import backtype.storm.tuple.Tuple;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Collections;
import java.util.List;
-public class FlattenEventMapper implements StreamEventMapper{
+public class FlattenEventMapper implements StreamEventMapper {
private final String streamId;
- private final static String TIMESTAMP_FIELD = "timestamp";
- private final static Logger LOGGER = LoggerFactory.getLogger(FlattenEventMapper.class);
+ private static final String TIMESTAMP_FIELD = "timestamp";
+ private static final Logger LOGGER = LoggerFactory.getLogger(FlattenEventMapper.class);
- public FlattenEventMapper(String streamId){
+ public FlattenEventMapper(String streamId) {
this.streamId = streamId;
}
@Override
public List<StreamEvent> map(Tuple tuple) throws Exception {
long timestamp;
- if(tuple.getFields().contains(TIMESTAMP_FIELD)) {
+ if (tuple.getFields().contains(TIMESTAMP_FIELD)) {
try {
timestamp = tuple.getLongByField("timestamp");
} catch (Exception ex) {
// if timestamp is not null
- LOGGER.error(ex.getMessage(),ex);
+ LOGGER.error(ex.getMessage(), ex);
timestamp = 0;
}
} else {
timestamp = System.currentTimeMillis();
}
Object[] values = new Object[tuple.getFields().size()];
- for(int i=0;i<tuple.getFields().size();i++){
+ for (int i = 0; i < tuple.getFields().size(); i++) {
values[i] = tuple.getValue(i);
}
StreamEvent event = new StreamEvent();
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/c752ed86/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/sink/KafkaStreamSink.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/sink/KafkaStreamSink.java b/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/sink/KafkaStreamSink.java
index 27848d9..2a03275 100644
--- a/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/sink/KafkaStreamSink.java
+++ b/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/sink/KafkaStreamSink.java
@@ -30,7 +30,7 @@ import java.util.Map;
import java.util.Properties;
public class KafkaStreamSink extends StormStreamSink<KafkaStreamSinkConfig> {
- private final static Logger LOG = LoggerFactory.getLogger(KafkaStreamSink.class);
+ private static final Logger LOG = LoggerFactory.getLogger(KafkaStreamSink.class);
private String topicId;
private Producer producer;
private KafkaStreamSinkConfig config;
@@ -54,11 +54,11 @@ public class KafkaStreamSink extends StormStreamSink<KafkaStreamSinkConfig> {
}
@Override
- protected void execute(Object key, Map event,BasicOutputCollector collector) {
+ protected void execute(Object key, Map event, BasicOutputCollector collector) {
try {
String output = new ObjectMapper().writeValueAsString(event);
producer.send(new KeyedMessage(this.topicId, key, output));
- }catch(Exception ex){
+ } catch (Exception ex) {
LOG.error(ex.getMessage(), ex);
collector.reportError(ex);
}
@@ -69,17 +69,17 @@ public class KafkaStreamSink extends StormStreamSink<KafkaStreamSinkConfig> {
ensureTopicCreated();
}
- private void ensureTopicCreated(){
- LOG.info("TODO: ensure kafka topic {} created",this.topicId);
+ private void ensureTopicCreated() {
+ LOG.info("TODO: ensure kafka topic {} created", this.topicId);
}
- private void ensureTopicDeleted(){
- LOG.info("TODO: ensure kafka topic {} deleted",this.topicId);
+ private void ensureTopicDeleted() {
+ LOG.info("TODO: ensure kafka topic {} deleted", this.topicId);
}
@Override
public void cleanup() {
- if(this.producer != null){
+ if (this.producer != null) {
this.producer.close();
}
}
@@ -89,7 +89,7 @@ public class KafkaStreamSink extends StormStreamSink<KafkaStreamSinkConfig> {
ensureTopicDeleted();
}
- public static class Provider implements StreamSinkProvider<KafkaStreamSink,KafkaStreamSinkConfig> {
+ public static class Provider implements StreamSinkProvider<KafkaStreamSink, KafkaStreamSinkConfig> {
@Override
public KafkaStreamSinkConfig getSinkConfig(String streamId, Config config) {
KafkaStreamSinkConfig desc = new KafkaStreamSinkConfig();
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/c752ed86/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/sink/LoggingStreamSink.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/sink/LoggingStreamSink.java b/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/sink/LoggingStreamSink.java
index 787fc99..61a6836 100644
--- a/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/sink/LoggingStreamSink.java
+++ b/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/sink/LoggingStreamSink.java
@@ -1,4 +1,4 @@
-/**
+/*
* 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.
@@ -24,7 +24,7 @@ import org.slf4j.LoggerFactory;
import java.util.Map;
public class LoggingStreamSink extends StormStreamSink<DefaultStreamSinkConfig> {
- private final static Logger LOGGER = LoggerFactory.getLogger(KafkaStreamSink.class);
+ private static final Logger LOGGER = LoggerFactory.getLogger(KafkaStreamSink.class);
@Override
public void onInstall() {
@@ -38,10 +38,10 @@ public class LoggingStreamSink extends StormStreamSink<DefaultStreamSinkConfig>
@Override
protected void execute(Object key, Map event, BasicOutputCollector collector) {
- LOGGER.info("Receiving {}",event);
+ LOGGER.info("Receiving {}", event);
}
- public static class Provider implements StreamSinkProvider<LoggingStreamSink,DefaultStreamSinkConfig> {
+ public static class Provider implements StreamSinkProvider<LoggingStreamSink, DefaultStreamSinkConfig> {
@Override
public DefaultStreamSinkConfig getSinkConfig(String streamId, Config config) {
return new DefaultStreamSinkConfig(LoggingStreamSink.class);
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/c752ed86/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/sink/StormStreamSink.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/sink/StormStreamSink.java b/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/sink/StormStreamSink.java
index d152e76..7ea234a 100644
--- a/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/sink/StormStreamSink.java
+++ b/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/sink/StormStreamSink.java
@@ -1,4 +1,4 @@
-/**
+/*
* 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.
@@ -16,26 +16,20 @@
*/
package org.apache.eagle.app.sink;
+import org.apache.eagle.metadata.model.StreamSinkConfig;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
-import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.google.common.base.Preconditions;
-import kafka.producer.KeyedMessage;
-import org.apache.eagle.alert.engine.model.StreamEvent;
-import org.apache.eagle.metadata.model.StreamSinkConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.HashMap;
-import java.util.List;
import java.util.Map;
public abstract class StormStreamSink<K extends StreamSinkConfig> extends BaseBasicBolt implements StreamSink<K> {
- private final static Logger LOG = LoggerFactory.getLogger(StormStreamSink.class);
+ private static final Logger LOG = LoggerFactory.getLogger(StormStreamSink.class);
private String streamId;
@Override
@@ -49,14 +43,14 @@ public abstract class StormStreamSink<K extends StreamSinkConfig> extends BaseBa
}
/**
- * Implicitly hides the Tuple protocol inside code as Tuple[Key,Map]
+ * Implicitly hides the Tuple protocol inside code as Tuple[Key,Map].
*/
@Override
public void execute(Tuple input, BasicOutputCollector collector) {
try {
Map event = null;
Object key = input.getValue(0);
- if(input.size()<2){
+ if (input.size() < 2) {
event = tupleAsMap(input);
} else {
Object value = input.getValue(1);
@@ -68,23 +62,23 @@ public abstract class StormStreamSink<K extends StreamSinkConfig> extends BaseBa
}
}
}
- execute(key,event,collector);
- }catch(Exception ex){
+ execute(key, event, collector);
+ } catch (Exception ex) {
LOG.error(ex.getMessage(), ex);
collector.reportError(ex);
}
}
- private Map tupleAsMap(Tuple tuple){
+ protected abstract void execute(Object key, Map event, BasicOutputCollector collector);
+
+ private Map tupleAsMap(Tuple tuple) {
Map values = new HashMap<>();
- for(String field:tuple.getFields()){
- values.put(field,tuple.getValueByField(field));
+ for (String field : tuple.getFields()) {
+ values.put(field, tuple.getValueByField(field));
}
return values;
}
- protected abstract void execute(Object key,Map event,BasicOutputCollector collector);
-
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
}
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/c752ed86/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/sink/StreamEventMapper.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/sink/StreamEventMapper.java b/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/sink/StreamEventMapper.java
index 350e274..710e4d2 100644
--- a/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/sink/StreamEventMapper.java
+++ b/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/sink/StreamEventMapper.java
@@ -16,15 +16,17 @@
*/
package org.apache.eagle.app.sink;
-import backtype.storm.tuple.Tuple;
import org.apache.eagle.alert.engine.model.StreamEvent;
+import backtype.storm.tuple.Tuple;
import java.io.Serializable;
import java.util.List;
@FunctionalInterface
-public interface StreamEventMapper extends Serializable{
+public interface StreamEventMapper extends Serializable {
/**
+ * Map from storm tuple to Stream Event.
+ *
* @param tuple
* @return
* @throws Exception
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/c752ed86/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/sink/StreamSinkProvider.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/sink/StreamSinkProvider.java b/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/sink/StreamSinkProvider.java
index 6fce77b..0425fa6 100644
--- a/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/sink/StreamSinkProvider.java
+++ b/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/sink/StreamSinkProvider.java
@@ -16,31 +16,27 @@
*/
package org.apache.eagle.app.sink;
-import com.typesafe.config.Config;
import org.apache.eagle.metadata.model.StreamSinkConfig;
+import com.typesafe.config.Config;
import java.lang.reflect.ParameterizedType;
-public interface StreamSinkProvider<S extends StreamSink<D>,D extends StreamSinkConfig>{
- /**
- * @param streamId
- * @param config
- * @return
- */
+public interface StreamSinkProvider<S extends StreamSink<D>, D extends StreamSinkConfig> {
D getSinkConfig(String streamId, Config config);
+
S getSink();
- default S getSink(String streamId, Config config){
+ default S getSink(String streamId, Config config) {
S s = getSink();
- s.init(streamId,getSinkConfig(streamId,config));
+ s.init(streamId, getSinkConfig(streamId, config));
return s;
}
- default Class<? extends S> getSinkType(){
+ default Class<? extends S> getSinkType() {
return (Class<S>) ((ParameterizedType) getClass().getGenericSuperclass()).getActualTypeArguments()[0];
}
- default Class<? extends D> getSinkConfigType(){
+ default Class<? extends D> getSinkConfigType() {
return (Class<D>) ((ParameterizedType) getClass().getGenericSuperclass()).getActualTypeArguments()[1];
}
}
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/c752ed86/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/spi/AbstractApplicationProvider.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/spi/AbstractApplicationProvider.java b/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/spi/AbstractApplicationProvider.java
index 6975b6b..4a9c1ff 100644
--- a/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/spi/AbstractApplicationProvider.java
+++ b/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/spi/AbstractApplicationProvider.java
@@ -16,7 +16,6 @@
*/
package org.apache.eagle.app.spi;
-import com.typesafe.config.Config;
import org.apache.eagle.alert.engine.coordinator.StreamDefinition;
import org.apache.eagle.app.Application;
import org.apache.eagle.app.config.ApplicationProviderConfig;
@@ -25,11 +24,12 @@ import org.apache.eagle.common.module.ModuleRegistry;
import org.apache.eagle.metadata.model.ApplicationDesc;
import org.apache.eagle.metadata.model.ApplicationDocs;
import org.apache.eagle.metadata.model.Configuration;
+import com.typesafe.config.Config;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import javax.xml.bind.JAXBException;
import java.util.List;
+import javax.xml.bind.JAXBException;
/**
* Default metadata path is: /META-INF/providers/${ApplicationProviderClassName}.xml
@@ -37,20 +37,20 @@ import java.util.List;
* @param <T>
*/
public abstract class AbstractApplicationProvider<T extends Application> implements ApplicationProvider<T> {
- private final static Logger LOG = LoggerFactory.getLogger(AbstractApplicationProvider.class);
+ private static final Logger LOG = LoggerFactory.getLogger(AbstractApplicationProvider.class);
private final ApplicationDesc applicationDesc;
- private final static String METADATA_RESOURCE_PATH="/META-INF/providers/%s.xml";
+ private static final String METADATA_RESOURCE_PATH = "/META-INF/providers/%s.xml";
/**
* Default metadata path is: /META-INF/providers/${ApplicationProviderClassName}.xml
*
- * You are not recommended to override this method except you could make sure the path is universal unique
+ * <p>You are not recommended to override this method except you could make sure the path is universal unique</p>
*
* @return metadata file path
*/
- protected final String getMetadata(){
- return String.format(METADATA_RESOURCE_PATH,this.getClass().getName());
+ protected final String getMetadata() {
+ return String.format(METADATA_RESOURCE_PATH, this.getClass().getName());
}
protected AbstractApplicationProvider() {
@@ -64,7 +64,7 @@ public abstract class AbstractApplicationProvider<T extends Application> impleme
setDocs(descWrapperConfig.getDocs());
try {
if (descWrapperConfig.getAppClass() != null) {
-// setAppClass((Class<T>) Class.forName(descWrapperConfig.getAppClass()));
+ // setAppClass((Class<T>) Class.forName(descWrapperConfig.getAppClass()));
setAppClass((Class<T>) Class.forName(descWrapperConfig.getAppClass(), true, this.getClass().getClassLoader()));
if (!Application.class.isAssignableFrom(applicationDesc.getAppClass())) {
throw new IllegalStateException(descWrapperConfig.getAppClass() + " is not sub-class of " + Application.class.getCanonicalName());
@@ -107,16 +107,16 @@ public abstract class AbstractApplicationProvider<T extends Application> impleme
try {
applicationDesc.setConfiguration(Configuration.fromResource(resourceName));
} catch (JAXBException e) {
- LOG.error("Failed to load configuration template from "+resourceName,e);
- throw new RuntimeException("Failed to load configuration template from "+resourceName,e);
+ LOG.error("Failed to load configuration template from " + resourceName, e);
+ throw new RuntimeException("Failed to load configuration template from " + resourceName, e);
}
}
@Override
public String toString() {
return String.format(
- "%s[name=%s, type=%s, version=%s, viewPath=%s, appClass=%s, configuration= %s properties]", getClass().getSimpleName(),
- applicationDesc.getName(),applicationDesc.getType(),applicationDesc.getVersion(), applicationDesc.getViewPath(), applicationDesc.getAppClass(), applicationDesc.getConfiguration().size()
+ "%s[name=%s, type=%s, version=%s, viewPath=%s, appClass=%s, configuration= %s properties]", getClass().getSimpleName(),
+ applicationDesc.getName(), applicationDesc.getType(), applicationDesc.getVersion(), applicationDesc.getViewPath(), applicationDesc.getAppClass(), applicationDesc.getConfiguration().size()
);
}
@@ -140,6 +140,6 @@ public abstract class AbstractApplicationProvider<T extends Application> impleme
@Override
public void register(ModuleRegistry registry) {
- LOG.debug("Registering modules {}",this.getClass().getName());
+ LOG.debug("Registering modules {}", this.getClass().getName());
}
}
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/c752ed86/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/spi/ApplicationProvider.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/spi/ApplicationProvider.java b/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/spi/ApplicationProvider.java
index 0c68cfc..761cfbb 100644
--- a/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/spi/ApplicationProvider.java
+++ b/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/spi/ApplicationProvider.java
@@ -16,23 +16,23 @@
*/
package org.apache.eagle.app.spi;
-import com.typesafe.config.Config;
import org.apache.eagle.app.Application;
import org.apache.eagle.app.config.ApplicationProviderConfig;
-import org.apache.eagle.metadata.model.ApplicationDesc;
import org.apache.eagle.common.module.ModuleRegistry;
+import org.apache.eagle.metadata.model.ApplicationDesc;
+import com.typesafe.config.Config;
public interface ApplicationProvider<T extends Application> {
void prepare(ApplicationProviderConfig providerConfig,Config envConfig);
/**
- * @return application descriptor
+ * @return application descriptor.
*/
ApplicationDesc getApplicationDesc();
/**
- * @return application instance
+ * @return application instance.
*/
T getApplication();
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/c752ed86/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/test/AppJUnitRunner.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/test/AppJUnitRunner.java b/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/test/AppJUnitRunner.java
deleted file mode 100644
index b8174bb..0000000
--- a/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/test/AppJUnitRunner.java
+++ /dev/null
@@ -1,87 +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.eagle.app.test;
-
-import com.google.inject.Guice;
-import com.google.inject.Injector;
-import com.google.inject.Module;
-import org.apache.eagle.common.module.Modules;
-import org.junit.runners.BlockJUnit4ClassRunner;
-import org.junit.runners.model.InitializationError;
-
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
-
-public class AppJUnitRunner extends BlockJUnit4ClassRunner {
- private final Injector injector;
- public AppJUnitRunner(Class<?> klass) throws InitializationError {
- super(klass);
- injector = createInjectorFor(getModulesFor(klass));
- }
-
- @Override
- protected Object createTest() throws Exception {
- final Object obj = super.createTest();
- injector.injectMembers(this);
- this.injector.injectMembers(obj);
- return obj;
- }
-
- /**
- * Create a Guice Injector for the class under test.
- * @param classes Guice Modules
- * @return A Guice Injector instance.
- * @throws InitializationError If couldn't instantiate a module.
- */
- private Injector createInjectorFor(final Class<?>[] classes)
- throws InitializationError {
- final List<Module> modules = new ArrayList<>();
-
- ApplicationTestGuiceModule testGuiceModule = new ApplicationTestGuiceModule();
-
- // Add default modules
- modules.add(testGuiceModule);
-
- if(classes!= null) {
- for (final Class<?> module : Arrays.asList(classes)) {
- try {
- modules.add((Module) module.newInstance());
- } catch (final ReflectiveOperationException exception) {
- throw new InitializationError(exception);
- }
- }
- }
- return Guice.createInjector(modules);
- }
-
- /**
- * Get the list of Guice Modules request by GuiceModules annotation in the
- * class under test.
- * @param klass Class under test.
- * @return A Class Array of Guice Modules required by this class.
- * @throws InitializationError If the annotation is not present.
- */
- private Class<?>[] getModulesFor(final Class<?> klass)
- throws InitializationError {
- final Modules annotation = klass.getAnnotation(Modules.class);
- if (annotation == null) {
- return null;
- }
- return annotation.value();
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/c752ed86/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/test/ApplicationSimulator.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/test/ApplicationSimulator.java b/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/test/ApplicationSimulator.java
index 3e4aa21..d8f342f 100644
--- a/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/test/ApplicationSimulator.java
+++ b/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/test/ApplicationSimulator.java
@@ -16,53 +16,35 @@
*/
package org.apache.eagle.app.test;
+import org.apache.eagle.app.spi.ApplicationProvider;
import com.google.inject.Guice;
import com.google.inject.Module;
-import org.apache.eagle.app.spi.ApplicationProvider;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
/**
- * Application test simulator for developer to quickly run application without diving into application lifecycle
+ * Application test simulator for developer to quickly run application without diving into application lifecycle.
*/
public abstract class ApplicationSimulator {
- /**
- *
- * @param appType
- */
public abstract void start(String appType);
- /**
- *
- * @param appType
- * @param appConfig
- */
- public abstract void start(String appType, Map<String,Object> appConfig);
+ public abstract void start(String appType, Map<String, Object> appConfig);
- /**
- *
- * @param appProviderClass
- */
public abstract void start(Class<? extends ApplicationProvider> appProviderClass);
- /**
- *
- * @param appProviderClass
- * @param appConfig
- */
- public abstract void start(Class<? extends ApplicationProvider> appProviderClass, Map<String,Object> appConfig) throws Exception;
+ public abstract void start(Class<? extends ApplicationProvider> appProviderClass, Map<String, Object> appConfig) throws Exception;
- public static ApplicationSimulator getInstance(){
+ public static ApplicationSimulator getInstance() {
return Guice.createInjector(new ApplicationTestGuiceModule()).getInstance(ApplicationSimulator.class);
}
/**
- * @param modules additional modules
- * @return ApplicationSimulator instance
+ * @param modules additional modules.
+ * @return ApplicationSimulator instance.
*/
- public static ApplicationSimulator getInstance(Module ... modules){
+ public static ApplicationSimulator getInstance(Module... modules) {
List<Module> contextModules = Arrays.asList(modules);
contextModules.add(new ApplicationTestGuiceModule());
return Guice.createInjector(contextModules).getInstance(ApplicationSimulator.class);
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/c752ed86/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/test/ApplicationSimulatorImpl.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/test/ApplicationSimulatorImpl.java b/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/test/ApplicationSimulatorImpl.java
index 35dead2..2094d74 100644
--- a/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/test/ApplicationSimulatorImpl.java
+++ b/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/test/ApplicationSimulatorImpl.java
@@ -16,8 +16,6 @@
*/
package org.apache.eagle.app.test;
-import com.google.inject.Inject;
-import com.typesafe.config.Config;
import org.apache.eagle.app.config.ApplicationProviderConfig;
import org.apache.eagle.app.resource.ApplicationResource;
import org.apache.eagle.app.service.ApplicationOperations;
@@ -26,6 +24,8 @@ import org.apache.eagle.app.utils.DynamicJarPathFinder;
import org.apache.eagle.metadata.model.ApplicationEntity;
import org.apache.eagle.metadata.model.SiteEntity;
import org.apache.eagle.metadata.resource.SiteResource;
+import com.google.inject.Inject;
+import com.typesafe.config.Config;
import org.junit.Assert;
import java.util.HashMap;
@@ -38,12 +38,23 @@ public class ApplicationSimulatorImpl extends ApplicationSimulator {
private final ApplicationResource applicationResource;
@Inject
- public ApplicationSimulatorImpl(Config config, SiteResource siteResource, ApplicationResource applicationResource){
+ public ApplicationSimulatorImpl(Config config, SiteResource siteResource, ApplicationResource applicationResource) {
this.config = config;
this.siteResource = siteResource;
this.applicationResource = applicationResource;
}
+ private static final AtomicInteger incr = new AtomicInteger();
+
+ private SiteEntity getUniqueSite() {
+ // Create local site
+ SiteEntity siteEntity = new SiteEntity();
+ siteEntity.setSiteId("SIMULATED_SITE_" + incr.incrementAndGet());
+ siteEntity.setSiteName(siteEntity.getSiteId());
+ siteEntity.setDescription("Automatically generated unique simulation site " + siteEntity.getSiteId() + " (simulator: " + this + ")");
+ return siteEntity;
+ }
+
@Override
public void start(String appType) {
start(appType, new HashMap<>());
@@ -54,26 +65,15 @@ public class ApplicationSimulatorImpl extends ApplicationSimulator {
SiteEntity siteEntity = getUniqueSite();
siteResource.createSite(siteEntity);
Assert.assertNotNull(siteEntity.getUuid());
- ApplicationOperations.InstallOperation installOperation = new ApplicationOperations.InstallOperation(siteEntity.getSiteId(),appType, ApplicationEntity.Mode.LOCAL);
+ ApplicationOperations.InstallOperation installOperation = new ApplicationOperations.InstallOperation(siteEntity.getSiteId(), appType, ApplicationEntity.Mode.LOCAL);
installOperation.setConfiguration(appConfig);
// Install application
ApplicationEntity applicationEntity =
- applicationResource.installApplication(installOperation).getData();
+ applicationResource.installApplication(installOperation).getData();
// Start application
applicationResource.startApplication(new ApplicationOperations.StartOperation(applicationEntity.getUuid()));
}
- private final static AtomicInteger incr = new AtomicInteger();
-
- private SiteEntity getUniqueSite(){
- // Create local site
- SiteEntity siteEntity = new SiteEntity();
- siteEntity.setSiteId("SIMULATED_SITE_"+incr.incrementAndGet());
- siteEntity.setSiteName(siteEntity.getSiteId());
- siteEntity.setDescription("Automatically generated unique simulation site "+siteEntity.getSiteId()+" (simulator: "+this+")");
- return siteEntity;
- }
-
@Override
public void start(Class<? extends ApplicationProvider> appProviderClass) {
start(appProviderClass, new HashMap<>());
@@ -83,10 +83,10 @@ public class ApplicationSimulatorImpl extends ApplicationSimulator {
public void start(Class<? extends ApplicationProvider> appProviderClass, Map<String, Object> appConfig) {
try {
ApplicationProvider applicationProvider = appProviderClass.newInstance();
- applicationProvider.prepare(new ApplicationProviderConfig(DynamicJarPathFinder.findPath(appProviderClass),appProviderClass),config);
- start(applicationProvider.getApplicationDesc().getType(),appConfig);
+ applicationProvider.prepare(new ApplicationProviderConfig(DynamicJarPathFinder.findPath(appProviderClass), appProviderClass), config);
+ start(applicationProvider.getApplicationDesc().getType(), appConfig);
} catch (InstantiationException | IllegalAccessException e) {
- throw new IllegalStateException(e.getMessage(),e);
+ throw new IllegalStateException(e.getMessage(), e);
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/c752ed86/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/test/ApplicationTestBase.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/test/ApplicationTestBase.java b/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/test/ApplicationTestBase.java
index 1c7d6be..67e93a8 100644
--- a/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/test/ApplicationTestBase.java
+++ b/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/test/ApplicationTestBase.java
@@ -17,27 +17,19 @@
package org.apache.eagle.app.test;
import com.google.inject.Guice;
-import com.google.inject.Inject;
import com.google.inject.Injector;
-import org.apache.commons.dbcp.BasicDataSource;
-import org.apache.eagle.common.module.CommonGuiceModule;
-import org.apache.eagle.metadata.service.memory.MemoryMetadataStore;
-import org.junit.After;
import org.junit.Before;
-import javax.sql.DataSource;
-import java.sql.SQLException;
-
public class ApplicationTestBase {
private Injector injector;
@Before
- public void setUp(){
+ public void setUp() {
injector = Guice.createInjector(new ApplicationTestGuiceModule());
injector.injectMembers(this);
}
- public Injector injector(){
+ public Injector injector() {
return injector;
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/c752ed86/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/test/ApplicationTestGuiceModule.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/test/ApplicationTestGuiceModule.java b/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/test/ApplicationTestGuiceModule.java
index b50a661..3e2ba2c 100644
--- a/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/test/ApplicationTestGuiceModule.java
+++ b/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/test/ApplicationTestGuiceModule.java
@@ -16,20 +16,19 @@
*/
package org.apache.eagle.app.test;
-import com.google.inject.AbstractModule;
-import com.google.inject.Singleton;
-import com.typesafe.config.ConfigFactory;
import org.apache.eagle.app.module.ApplicationExtensionLoader;
import org.apache.eagle.app.module.ApplicationGuiceModule;
import org.apache.eagle.app.service.ApplicationProviderService;
import org.apache.eagle.app.service.impl.ApplicationProviderServiceImpl;
-import org.apache.eagle.app.spi.ApplicationProvider;
import org.apache.eagle.common.module.CommonGuiceModule;
import org.apache.eagle.common.module.GlobalScope;
import org.apache.eagle.common.module.ModuleRegistry;
import org.apache.eagle.metadata.service.memory.MemoryMetadataStore;
+import com.google.inject.AbstractModule;
+import com.google.inject.Singleton;
+import com.typesafe.config.ConfigFactory;
-public class ApplicationTestGuiceModule extends AbstractModule{
+public class ApplicationTestGuiceModule extends AbstractModule {
@Override
protected void configure() {
CommonGuiceModule common = new CommonGuiceModule();
@@ -39,7 +38,7 @@ public class ApplicationTestGuiceModule extends AbstractModule{
install(common);
install(app);
install(store);
- ModuleRegistry registry =ApplicationExtensionLoader.load(common,app,store);
+ ModuleRegistry registry = ApplicationExtensionLoader.load(common, app, store);
registry.getModules(store.getClass()).forEach(this::install);
registry.getModules(GlobalScope.class).forEach(this::install);
bind(ApplicationSimulator.class).to(ApplicationSimulatorImpl.class).in(Singleton.class);
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/c752ed86/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/utils/DynamicJarPathFinder.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/utils/DynamicJarPathFinder.java b/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/utils/DynamicJarPathFinder.java
index a0a7d50..5e20b25 100644
--- a/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/utils/DynamicJarPathFinder.java
+++ b/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/utils/DynamicJarPathFinder.java
@@ -30,7 +30,8 @@ import java.nio.charset.Charset;
* https://github.com/rzwitserloot/lombok.patcher/blob/master/src/lombok/patcher/inject/LiveInjector.java
*/
public class DynamicJarPathFinder {
- private final static Logger LOG = LoggerFactory.getLogger(DynamicJarPathFinder.class);
+ private static final Logger LOG = LoggerFactory.getLogger(DynamicJarPathFinder.class);
+
/**
* If the provided class has been loaded from a jar file that is on the local file system, will find the absolute path to that jar file.
*
@@ -40,12 +41,15 @@ public class DynamicJarPathFinder {
* other custom classloading device).
*/
public static String findPathJar(Class<?> context) throws IllegalStateException {
- if (context == null) context = DynamicJarPathFinder.class;
+ if (context == null) {
+ context = DynamicJarPathFinder.class;
+ }
String rawName = context.getName();
String classFileName;
- /* rawName is something like package.name.ContainingClass$ClassName. We need to turn this into ContainingClass$ClassName.class. */ {
+ /* rawName is something like package.name.ContainingClass$ClassName. We need to turn this into ContainingClass$ClassName.class. */
+ {
int idx = rawName.lastIndexOf('.');
- classFileName = (idx == -1 ? rawName : rawName.substring(idx+1)) + ".class";
+ classFileName = (idx == -1 ? rawName : rawName.substring(idx + 1)) + ".class";
}
String uri = context.getResource(classFileName).toString();
@@ -55,13 +59,15 @@ public class DynamicJarPathFinder {
if (!uri.startsWith("jar:file:")) {
int idx = uri.indexOf(':');
String protocol = idx == -1 ? "(unknown)" : uri.substring(0, idx);
- throw new IllegalStateException("This class has been loaded remotely via the " + protocol +
- " protocol. Only loading from a jar on the local file system is supported.");
+ throw new IllegalStateException("This class has been loaded remotely via the " + protocol
+ + " protocol. Only loading from a jar on the local file system is supported.");
}
int idx = uri.indexOf('!');
//As far as I know, the if statement below can't ever trigger, so it's more of a sanity check thing.
- if (idx == -1) throw new IllegalStateException("You appear to have loaded this class from a local jar file, but I can't make sense of the URL!");
+ if (idx == -1) {
+ throw new IllegalStateException("You appear to have loaded this class from a local jar file, but I can't make sense of the URL!");
+ }
try {
String fileName = URLDecoder.decode(uri.substring("jar:file:".length(), idx), Charset.defaultCharset().name());
@@ -74,21 +80,24 @@ public class DynamicJarPathFinder {
/**
* Similar to JarPathFinder, but not make sure the path must valid jar.
*
- * @see DynamicJarPathFinder#findPathJar(Class)
* @return the class path contains the context class
+ * @see DynamicJarPathFinder#findPathJar(Class)
*/
public static String findPath(Class<?> context) throws IllegalStateException {
- if (context == null) context = DynamicJarPathFinder.class;
+ if (context == null) {
+ context = DynamicJarPathFinder.class;
+ }
String rawName = context.getName();
String classFileName;
- /* rawName is something like package.name.ContainingClass$ClassName. We need to turn this into ContainingClass$ClassName.class. */ {
+ /* rawName is something like package.name.ContainingClass$ClassName. We need to turn this into ContainingClass$ClassName.class. */
+ {
int idx = rawName.lastIndexOf('.');
- classFileName = (idx == -1 ? rawName : rawName.substring(idx+1)) + ".class";
+ classFileName = (idx == -1 ? rawName : rawName.substring(idx + 1)) + ".class";
}
String uri = context.getResource(classFileName).toString();
if (uri.startsWith("file:")) {
- LOG.warn("This class has been loaded from a directory and not from a jar file: {}",uri);
+ LOG.warn("This class has been loaded from a directory and not from a jar file: {}", uri);
String fileName = null;
try {
fileName = URLDecoder.decode(uri.substring("file:".length(), uri.length()), Charset.defaultCharset().name());
@@ -101,8 +110,8 @@ public class DynamicJarPathFinder {
if (!uri.startsWith("jar:file:")) {
int idx = uri.indexOf(':');
String protocol = idx == -1 ? "(unknown)" : uri.substring(0, idx);
- throw new IllegalStateException("This class has been loaded remotely via the " + protocol +
- " protocol. Only loading from a jar on the local file system is supported.");
+ throw new IllegalStateException("This class has been loaded remotely via the " + protocol
+ + " protocol. Only loading from a jar on the local file system is supported.");
}
int idx = uri.indexOf('!');
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/c752ed86/eagle-core/eagle-app/pom.xml
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-app/pom.xml b/eagle-core/eagle-app/pom.xml
index c2bfd97..5aff6a6 100644
--- a/eagle-core/eagle-app/pom.xml
+++ b/eagle-core/eagle-app/pom.xml
@@ -34,4 +34,17 @@
<modules>
<module>eagle-app-base</module>
</modules>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-checkstyle-plugin</artifactId>
+ <configuration>
+ <failOnViolation>true</failOnViolation>
+ <failsOnError>true</failsOnError>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
</project>
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/c752ed86/eagle-core/eagle-common/src/main/java/org/apache/eagle/common/config/EagleConfig.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-common/src/main/java/org/apache/eagle/common/config/EagleConfig.java b/eagle-core/eagle-common/src/main/java/org/apache/eagle/common/config/EagleConfig.java
index 6226057..8b4b959 100755
--- a/eagle-core/eagle-common/src/main/java/org/apache/eagle/common/config/EagleConfig.java
+++ b/eagle-core/eagle-common/src/main/java/org/apache/eagle/common/config/EagleConfig.java
@@ -16,9 +16,9 @@
*/
package org.apache.eagle.common.config;
-import com.typesafe.config.Config;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.client.HTableInterface;
+import com.typesafe.config.Config;
import java.util.TimeZone;
import java.util.concurrent.ThreadPoolExecutor;
@@ -27,7 +27,7 @@ public interface EagleConfig {
boolean isCoprocessorEnabled();
- HTableInterface getHTable(String tableName);
+ HTableInterface getHTable(String tableName);
Configuration getHbaseConf();
@@ -35,26 +35,26 @@ public interface EagleConfig {
ThreadPoolExecutor getExecutor();
- String getZKQuorum();
+ String getZKQuorum();
- String getZKPort();
+ String getZKPort();
- String getServiceHost();
+ String getServiceHost();
- int getServicePort();
+ int getServicePort();
String getEnv();
boolean isTableNamePrefixedWithEnvironment();
-
+
int getHBaseClientScanCacheSize();
TimeZone getTimeZone();
-
+
boolean isServiceAuditingEnabled();
/**
- * @return root config
+ * @return root config.
*/
Config getConfig();
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/c752ed86/eagle-core/eagle-common/src/main/java/org/apache/eagle/common/config/EagleConfigConstants.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-common/src/main/java/org/apache/eagle/common/config/EagleConfigConstants.java b/eagle-core/eagle-common/src/main/java/org/apache/eagle/common/config/EagleConfigConstants.java
index 7829728..693041d 100644
--- a/eagle-core/eagle-common/src/main/java/org/apache/eagle/common/config/EagleConfigConstants.java
+++ b/eagle-core/eagle-common/src/main/java/org/apache/eagle/common/config/EagleConfigConstants.java
@@ -17,49 +17,49 @@
package org.apache.eagle.common.config;
public final class EagleConfigConstants {
- public final static String SERVICE_ENV = "service.env";
- public final static String SERVICE_HOST = "service.host";
- public final static String SERVICE_PORT = "service.port";
- public final static String SERVICE_HBASE_ZOOKEEPER_QUORUM = "storage.hbase.hbase-zookeeper-quorum";
- public final static String SERVICE_HBASE_ZOOKEEPER_PROPERTY_CLIENTPORT = "storage.hbase-zookeeper-property-clientPort";
- public final static String SERVICE_ZOOKEEPER_ZNODE_PARENT = "storage.zookeeper-znode-parent";
- public final static String SERVICE_HBASE_CLIENT_IPC_POOL_SIZE = "storage.hbase-client-ipc-pool-size";
- public final static String SERVICE_STORAGE_TYPE = "storage.storage-type";
- public final static String SERVICE_COPROCESSOR_ENABLED = "storage.coprocessor-enabled";
- public final static String SERVICE_TABLE_NAME_PREFIXED_WITH_ENVIRONMENT = "storage.table-name-prefixed-with-environment";
- public final static String SERVICE_HBASE_CLIENT_SCAN_CACHE_SIZE = "storage.hbase-client-scan-cache-size";
- public final static String SERVICE_THREADPOOL_CORE_SIZE = "storage.threadpool-core-size";
- public final static String SERVICE_THREADPOOL_MAX_SIZE = "storage.threadpool-max-size";
- public final static String SERVICE_THREADPOOL_SHRINK_SIZE = "storage.threadpool-shrink-size";
- public final static String SERVICE_AUDITING_ENABLED = "storage.audit-enabled";
+ public static final String SERVICE_ENV = "service.env";
+ public static final String SERVICE_HOST = "service.host";
+ public static final String SERVICE_PORT = "service.port";
+ public static final String SERVICE_HBASE_ZOOKEEPER_QUORUM = "storage.hbase.zookeeperQuorum";
+ public static final String SERVICE_HBASE_ZOOKEEPER_PROPERTY_CLIENTPORT = "storage.hbase.zookeeperPropertyClientPort";
+ public static final String SERVICE_ZOOKEEPER_ZNODE_PARENT = "storage.hbase.zookeeperZnodeParent";
+ public static final String SERVICE_HBASE_CLIENT_IPC_POOL_SIZE = "storage.hbase.clientIpcPoolSize";
+ public static final String SERVICE_STORAGE_TYPE = "storage.type";
+ public static final String SERVICE_COPROCESSOR_ENABLED = "storage.hbase.coprocessorEnabled";
+ public static final String SERVICE_TABLE_NAME_PREFIXED_WITH_ENVIRONMENT = "storage.hbase.tableNamePrefixedWithEnvironment";
+ public static final String SERVICE_HBASE_CLIENT_SCAN_CACHE_SIZE = "storage.hbase.clientScanCacheSize";
+ public static final String SERVICE_THREADPOOL_CORE_SIZE = "storage.hbase.threadpoolCoreSize";
+ public static final String SERVICE_THREADPOOL_MAX_SIZE = "storage.hbase.threadpoolMaxSize";
+ public static final String SERVICE_THREADPOOL_SHRINK_SIZE = "storage.hbase.threadpoolShrinkSize";
+ public static final String SERVICE_AUDITING_ENABLED = "storage.hbase.auditEnabled";
- public final static String EAGLE_TIME_ZONE = "eagle.timezone";
- public final static String DEFAULT_EAGLE_TIME_ZONE = "UTC";
+ public static final String EAGLE_TIME_ZONE = "service.timezone";
+ public static final String DEFAULT_EAGLE_TIME_ZONE = "UTC";
- public final static int DEFAULT_THREAD_POOL_CORE_SIZE = 10;
- public final static int DEFAULT_THREAD_POOL_MAX_SIZE = 20;
- public final static long DEFAULT_THREAD_POOL_SHRINK_TIME = 60000L;
- public final static String DEFAULT_SERVICE_HOST = "localhost";
- public final static String DEFAULT_STORAGE_TYPE = "hbase";
- public final static int DEFAULT_SERVICE_PORT = 8080;
- public final static String DEFAULT_ZOOKEEPER_ZNODE_PARENT = "/hbase-unsecure";
+ public static final int DEFAULT_THREAD_POOL_CORE_SIZE = 10;
+ public static final int DEFAULT_THREAD_POOL_MAX_SIZE = 20;
+ public static final long DEFAULT_THREAD_POOL_SHRINK_TIME = 60000L;
+ public static final String DEFAULT_SERVICE_HOST = "localhost";
+ public static final String DEFAULT_STORAGE_TYPE = "hbase";
+ public static final int DEFAULT_SERVICE_PORT = 8080;
+ public static final String DEFAULT_ZOOKEEPER_ZNODE_PARENT = "/hbase-unsecure";
- public final static String EAGLE_PROPS="eagleProps";
- public final static String EAGLE_SERVICE = "eagleService";
- public final static String HOST = "host";
- public final static String PORT = "port";
- public final static String USERNAME = "username";
- public final static String PASSWORD = "password";
+ public static final String EAGLE_PROPS="eagleProps";
+ public static final String EAGLE_SERVICE = "eagleService";
+ public static final String HOST = "host";
+ public static final String PORT = "port";
+ public static final String USERNAME = "username";
+ public static final String PASSWORD = "password";
- public final static String SITE = "site";
+ public static final String SITE = "site";
@Deprecated
- public final static String DATA_SOURCE = "dataSource";
- public final static String APPLICATION = "application";
+ public static final String DATA_SOURCE = "dataSource";
+ public static final String APPLICATION = "application";
- public final static String WEB_CONFIG = "web";
- public final static String APP_CONFIG = "app";
- public final static String CLASSIFICATION_CONFIG = "classification";
+ public static final String WEB_CONFIG = "web";
+ public static final String APP_CONFIG = "app";
+ public static final String CLASSIFICATION_CONFIG = "classification";
- public final static String LOCAL_MODE = "local";
- public final static String CLUSTER_MODE = "cluster";
+ public static final String LOCAL_MODE = "local";
+ public static final String CLUSTER_MODE = "cluster";
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/c752ed86/eagle-core/eagle-common/src/main/java/org/apache/eagle/common/config/EagleConfigFactory.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-common/src/main/java/org/apache/eagle/common/config/EagleConfigFactory.java b/eagle-core/eagle-common/src/main/java/org/apache/eagle/common/config/EagleConfigFactory.java
index 0d73743..2bd144d 100755
--- a/eagle-core/eagle-common/src/main/java/org/apache/eagle/common/config/EagleConfigFactory.java
+++ b/eagle-core/eagle-common/src/main/java/org/apache/eagle/common/config/EagleConfigFactory.java
@@ -31,53 +31,53 @@ import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public class EagleConfigFactory implements EagleConfig {
- private static final Logger LOG = LoggerFactory.getLogger(EagleConfigFactory.class);
+ private static final Logger LOG = LoggerFactory.getLogger(EagleConfigFactory.class);
- private String env;
- private String zkQuorum;
- private String zkPort;
+ private String env;
+ private String zkQuorum;
+ private String zkPort;
private Configuration hbaseConf;
- private String eagleServiceHost;
- private int eagleServicePort;
+ private String eagleServiceHost;
+ private int eagleServicePort;
private String storageType;
private Config config;
private TimeZone timeZone;
public boolean isCoprocessorEnabled() {
- return isCoprocessorEnabled;
- }
+ return isCoprocessorEnabled;
+ }
+
+ private boolean isCoprocessorEnabled;
+
+ private boolean tableNamePrefixedWithEnv;
- private boolean isCoprocessorEnabled;
+ private HTablePool pool;
+ private int hbaseClientScanCacheSize = 1000;
- private boolean tableNamePrefixedWithEnv;
+ private ThreadPoolExecutor executor = null;
- private HTablePool pool;
- private int hbaseClientScanCacheSize = 1000;
+ private static EagleConfigFactory manager = new EagleConfigFactory();
- private ThreadPoolExecutor executor = null;
+ private EagleConfigFactory() {
+ init();
+ if (this.getStorageType() == null || this.getStorageType().equalsIgnoreCase("hbase")) {
+ this.pool = new HTablePool(this.hbaseConf, 10);
+ }
+ }
- private static EagleConfigFactory manager = new EagleConfigFactory();
+ public static EagleConfig load() {
+ return manager;
+ }
- private EagleConfigFactory(){
- init();
- if(this.getStorageType() == null || this.getStorageType().equalsIgnoreCase("hbase")) {
- this.pool = new HTablePool(this.hbaseConf, 10);
- }
- }
-
- public static EagleConfig load(){
- return manager;
- }
-
- public HTableInterface getHTable(String tableName){
+ public HTableInterface getHTable(String tableName) {
return pool.getTable(tableName);
}
- private String getString(Config config,String path,String defaultValue){
- if(config.hasPath(path)){
+ private String getString(Config config, String path, String defaultValue) {
+ if (config.hasPath(path)) {
return config.getString(path);
- }else{
+ } else {
return defaultValue;
}
}
@@ -90,93 +90,100 @@ public class EagleConfigFactory implements EagleConfig {
return executor;
}
- private void init(){
+ private void init() {
this.config = ConfigFactory.load();
- this.timeZone = TimeZone.getTimeZone((config.hasPath(EagleConfigConstants.EAGLE_TIME_ZONE)? config.getString(EagleConfigConstants.EAGLE_TIME_ZONE): EagleConfigConstants.DEFAULT_EAGLE_TIME_ZONE));
- this.env = config.hasPath(EagleConfigConstants.SERVICE_ENV) ? config.getString(EagleConfigConstants.SERVICE_ENV):"dev";
- this.zkQuorum = config.hasPath(EagleConfigConstants.SERVICE_HBASE_ZOOKEEPER_QUORUM) ? config.getString(EagleConfigConstants.SERVICE_HBASE_ZOOKEEPER_QUORUM):null;
- this.zkPort = config.hasPath(EagleConfigConstants.SERVICE_HBASE_ZOOKEEPER_PROPERTY_CLIENTPORT) ? config.getString(EagleConfigConstants.SERVICE_HBASE_ZOOKEEPER_PROPERTY_CLIENTPORT): null;
- String zkZnodeParent = config.hasPath(EagleConfigConstants.SERVICE_ZOOKEEPER_ZNODE_PARENT)? config.getString(EagleConfigConstants.SERVICE_ZOOKEEPER_ZNODE_PARENT): EagleConfigConstants.DEFAULT_ZOOKEEPER_ZNODE_PARENT;
- String clientIPCPoolSize = getString(config, EagleConfigConstants.SERVICE_HBASE_CLIENT_IPC_POOL_SIZE, "10");
- this.hbaseConf = HBaseConfiguration.create();
-
- if (this.zkQuorum != null)
+ this.timeZone = TimeZone.getTimeZone((config.hasPath(EagleConfigConstants.EAGLE_TIME_ZONE)
+ ? config.getString(EagleConfigConstants.EAGLE_TIME_ZONE) : EagleConfigConstants.DEFAULT_EAGLE_TIME_ZONE));
+ this.env = config.hasPath(EagleConfigConstants.SERVICE_ENV) ? config.getString(EagleConfigConstants.SERVICE_ENV) : "dev";
+ this.zkQuorum = config.hasPath(EagleConfigConstants.SERVICE_HBASE_ZOOKEEPER_QUORUM)
+ ? config.getString(EagleConfigConstants.SERVICE_HBASE_ZOOKEEPER_QUORUM) : null;
+ this.zkPort = config.hasPath(EagleConfigConstants.SERVICE_HBASE_ZOOKEEPER_PROPERTY_CLIENTPORT)
+ ? config.getString(EagleConfigConstants.SERVICE_HBASE_ZOOKEEPER_PROPERTY_CLIENTPORT) : null;
+ final String zkZnodeParent = config.hasPath(EagleConfigConstants.SERVICE_ZOOKEEPER_ZNODE_PARENT)
+ ? config.getString(EagleConfigConstants.SERVICE_ZOOKEEPER_ZNODE_PARENT) : EagleConfigConstants.DEFAULT_ZOOKEEPER_ZNODE_PARENT;
+ String clientIPCPoolSize = getString(config, EagleConfigConstants.SERVICE_HBASE_CLIENT_IPC_POOL_SIZE, "10");
+ this.hbaseConf = HBaseConfiguration.create();
+
+ if (this.zkQuorum != null) {
this.hbaseConf.set("hbase.zookeeper.quorum", this.zkQuorum);
+ }
- if (this.zkPort != null)
+ if (this.zkPort != null) {
this.hbaseConf.set("hbase.zookeeper.property.clientPort", this.zkPort);
+ }
- if(zkZnodeParent != null)
+ if (zkZnodeParent != null) {
this.hbaseConf.set("zookeeper.znode.parent", zkZnodeParent);
- else
+ } else {
this.hbaseConf.set("zookeeper.znode.parent", EagleConfigConstants.DEFAULT_ZOOKEEPER_ZNODE_PARENT);
+ }
this.hbaseConf.set("hbase.client.ipc.pool.size", clientIPCPoolSize);
- this.eagleServiceHost = config.hasPath(EagleConfigConstants.SERVICE_HOST) ? config.getString(EagleConfigConstants.SERVICE_HOST) : EagleConfigConstants.DEFAULT_SERVICE_HOST;
+ this.eagleServiceHost = config.hasPath(EagleConfigConstants.SERVICE_HOST) ? config.getString(EagleConfigConstants.SERVICE_HOST) : EagleConfigConstants.DEFAULT_SERVICE_HOST;
this.storageType = config.hasPath(EagleConfigConstants.SERVICE_STORAGE_TYPE) ? config.getString(EagleConfigConstants.SERVICE_STORAGE_TYPE) : EagleConfigConstants.DEFAULT_STORAGE_TYPE;
this.isCoprocessorEnabled = config.hasPath(EagleConfigConstants.SERVICE_COPROCESSOR_ENABLED) && config.getBoolean(EagleConfigConstants.SERVICE_COPROCESSOR_ENABLED);
- this.eagleServicePort = config.hasPath(EagleConfigConstants.SERVICE_PORT) ? config.getInt(EagleConfigConstants.SERVICE_PORT) : EagleConfigConstants.DEFAULT_SERVICE_PORT;
+ this.eagleServicePort = config.hasPath(EagleConfigConstants.SERVICE_PORT) ? config.getInt(EagleConfigConstants.SERVICE_PORT) : EagleConfigConstants.DEFAULT_SERVICE_PORT;
this.tableNamePrefixedWithEnv = config.hasPath(EagleConfigConstants.SERVICE_TABLE_NAME_PREFIXED_WITH_ENVIRONMENT) && config.getBoolean(EagleConfigConstants.SERVICE_TABLE_NAME_PREFIXED_WITH_ENVIRONMENT);
- this.hbaseClientScanCacheSize = config.hasPath(EagleConfigConstants.SERVICE_HBASE_CLIENT_SCAN_CACHE_SIZE)? config.getInt(EagleConfigConstants.SERVICE_HBASE_CLIENT_SCAN_CACHE_SIZE) : hbaseClientScanCacheSize;
+ this.hbaseClientScanCacheSize = config.hasPath(EagleConfigConstants.SERVICE_HBASE_CLIENT_SCAN_CACHE_SIZE) ? config.getInt(EagleConfigConstants.SERVICE_HBASE_CLIENT_SCAN_CACHE_SIZE) : hbaseClientScanCacheSize;
// initilize eagle service thread pool for parallel execution of hbase scan etc.
- int threadPoolCoreSize = config.hasPath(EagleConfigConstants.SERVICE_THREADPOOL_CORE_SIZE)? config.getInt(EagleConfigConstants.SERVICE_THREADPOOL_CORE_SIZE): EagleConfigConstants.DEFAULT_THREAD_POOL_CORE_SIZE;
- int threadPoolMaxSize = config.hasPath(EagleConfigConstants.SERVICE_THREADPOOL_MAX_SIZE) ? config.getInt(EagleConfigConstants.SERVICE_THREADPOOL_MAX_SIZE) : EagleConfigConstants.DEFAULT_THREAD_POOL_MAX_SIZE;
- long threadPoolShrinkTime = config.hasPath(EagleConfigConstants.SERVICE_THREADPOOL_SHRINK_SIZE) ? config.getLong(EagleConfigConstants.SERVICE_THREADPOOL_SHRINK_SIZE) : EagleConfigConstants.DEFAULT_THREAD_POOL_SHRINK_TIME;
- this.isServiceAuditingEnabled = config.hasPath(EagleConfigConstants.SERVICE_AUDITING_ENABLED) && config.getBoolean(EagleConfigConstants.SERVICE_AUDITING_ENABLED);
-
- this.executor = new ThreadPoolExecutor(threadPoolCoreSize, threadPoolMaxSize, threadPoolShrinkTime, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>());
-
- LOG.info("Successfully initialized config");
-
- if(LOG.isDebugEnabled()) {
- if(this.isCoprocessorEnabled){
- LOG.debug("Eagle HBase Coprocessor is enabled");
- }else{
- LOG.debug("Eagle HBase Coprocessor is disabled");
- }
- }
- }
+ int threadPoolCoreSize = config.hasPath(EagleConfigConstants.SERVICE_THREADPOOL_CORE_SIZE) ? config.getInt(EagleConfigConstants.SERVICE_THREADPOOL_CORE_SIZE) : EagleConfigConstants.DEFAULT_THREAD_POOL_CORE_SIZE;
+ int threadPoolMaxSize = config.hasPath(EagleConfigConstants.SERVICE_THREADPOOL_MAX_SIZE) ? config.getInt(EagleConfigConstants.SERVICE_THREADPOOL_MAX_SIZE) : EagleConfigConstants.DEFAULT_THREAD_POOL_MAX_SIZE;
+ long threadPoolShrinkTime = config.hasPath(EagleConfigConstants.SERVICE_THREADPOOL_SHRINK_SIZE) ? config.getLong(EagleConfigConstants.SERVICE_THREADPOOL_SHRINK_SIZE) : EagleConfigConstants.DEFAULT_THREAD_POOL_SHRINK_TIME;
+ this.isServiceAuditingEnabled = config.hasPath(EagleConfigConstants.SERVICE_AUDITING_ENABLED) && config.getBoolean(EagleConfigConstants.SERVICE_AUDITING_ENABLED);
+
+ this.executor = new ThreadPoolExecutor(threadPoolCoreSize, threadPoolMaxSize, threadPoolShrinkTime, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>());
+
+ LOG.info("Successfully initialized config");
+
+ if (LOG.isDebugEnabled()) {
+ if (this.isCoprocessorEnabled) {
+ LOG.debug("Eagle HBase Coprocessor is enabled");
+ } else {
+ LOG.debug("Eagle HBase Coprocessor is disabled");
+ }
+ }
+ }
@Override
- public Configuration getHbaseConf() {
- return hbaseConf;
- }
+ public Configuration getHbaseConf() {
+ return hbaseConf;
+ }
@Override
- public String getZKQuorum(){
- return this.zkQuorum;
+ public String getZKQuorum() {
+ return this.zkQuorum;
}
@Override
- public String getZKPort(){
- return this.zkPort;
- }
+ public String getZKPort() {
+ return this.zkPort;
+ }
@Override
- public String getServiceHost() {
- return eagleServiceHost;
- }
+ public String getServiceHost() {
+ return eagleServiceHost;
+ }
@Override
- public int getServicePort() {
- return eagleServicePort;
- }
+ public int getServicePort() {
+ return eagleServicePort;
+ }
@Override
- public String getEnv() {
- return env;
- }
+ public String getEnv() {
+ return env;
+ }
@Override
- public boolean isTableNamePrefixedWithEnvironment(){
- return this.tableNamePrefixedWithEnv;
- }
+ public boolean isTableNamePrefixedWithEnvironment() {
+ return this.tableNamePrefixedWithEnv;
+ }
@Override
- public int getHBaseClientScanCacheSize(){
- return this.hbaseClientScanCacheSize;
- }
+ public int getHBaseClientScanCacheSize() {
+ return this.hbaseClientScanCacheSize;
+ }
@Override
public TimeZone getTimeZone() {
@@ -187,12 +194,12 @@ public class EagleConfigFactory implements EagleConfig {
public Config getConfig() {
return this.config;
}
-
+
// added for jira EAGLE-47
boolean isServiceAuditingEnabled;
-
+
@Override
- public boolean isServiceAuditingEnabled() {
- return isServiceAuditingEnabled;
- }
+ public boolean isServiceAuditingEnabled() {
+ return isServiceAuditingEnabled;
+ }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/c752ed86/eagle-core/eagle-common/src/test/java/org/apache/eagle/common/config/TestEagleConfig.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-common/src/test/java/org/apache/eagle/common/config/TestEagleConfig.java b/eagle-core/eagle-common/src/test/java/org/apache/eagle/common/config/TestEagleConfig.java
index 892db0c..7e96433 100644
--- a/eagle-core/eagle-common/src/test/java/org/apache/eagle/common/config/TestEagleConfig.java
+++ b/eagle-core/eagle-common/src/test/java/org/apache/eagle/common/config/TestEagleConfig.java
@@ -17,27 +17,26 @@
package org.apache.eagle.common.config;
import org.junit.Assert;
-import org.junit.Before;
import org.junit.Test;
+import java.util.TimeZone;
+
/**
* @since 9/22/15
*/
public class TestEagleConfig {
- private EagleConfig config;
-
- @Before
- public void setUp(){
- System.setProperty("config.resource","test-service-config.conf");
- config = EagleConfigFactory.load();
- }
@Test
- public void testInit(){
+ public void testLoadConfig(){
+ System.setProperty("config.resource", "application-test.conf");
+ EagleConfig config = EagleConfigFactory.load();
Assert.assertEquals("test",config.getEnv());
- Assert.assertEquals("localhost",config.getServiceHost());
- Assert.assertEquals(9090,config.getServicePort());
+ Assert.assertEquals("localhost-for-test",config.getZKQuorum());
+ Assert.assertEquals("1234",config.getZKPort());
Assert.assertEquals("hbase",config.getStorageType());
- Assert.assertEquals(false,config.isCoprocessorEnabled());
+ Assert.assertEquals(true,config.isCoprocessorEnabled());
+ Assert.assertEquals(9090,config.getServicePort());
+ Assert.assertEquals("localhost",config.getServiceHost());
+ Assert.assertEquals(TimeZone.getTimeZone("UTC"),config.getTimeZone());
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/c752ed86/eagle-core/eagle-common/src/test/resources/application-test.conf
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-common/src/test/resources/application-test.conf b/eagle-core/eagle-common/src/test/resources/application-test.conf
new file mode 100644
index 0000000..587ec90
--- /dev/null
+++ b/eagle-core/eagle-common/src/test/resources/application-test.conf
@@ -0,0 +1,33 @@
+# 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.
+
+
+service {
+ timezone = "UTC"
+ env = "test"
+ host = "localhost"
+ port = 9090
+}
+
+storage{
+ type = "hbase"
+ hbase {
+ tableNamePrefixedwithEnvironment = true
+ zookeeperQuorum = "localhost-for-test"
+ zookeeperPropertyClientPort = 1234
+ zookeeperZnodeParent = "/hbase-for-test"
+ coprocessorEnabled = true
+ }
+}
\ No newline at end of file