eagle-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From h..@apache.org
Subject [12/18] incubator-eagle git commit: [EAGLE-530] Fix checkstyle problems on eagle-alert module and enable failOnViolation
Date Thu, 08 Sep 2016 07:14:13 GMT
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/a0fc8708/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/absence/AbsenceWindowProcessor.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/absence/AbsenceWindowProcessor.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/absence/AbsenceWindowProcessor.java
index 4e8d381..8f00b31 100644
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/absence/AbsenceWindowProcessor.java
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/absence/AbsenceWindowProcessor.java
@@ -33,65 +33,66 @@ public class AbsenceWindowProcessor {
     private boolean expired; // to mark if the time range has been went through
     private OccurStatus status = OccurStatus.not_sure;
 
-    public enum OccurStatus{
+    public enum OccurStatus {
         not_sure,
         occured,
         absent
     }
 
-    public AbsenceWindowProcessor(List<Object> expectAttrs, AbsenceWindow window){
+    public AbsenceWindowProcessor(List<Object> expectAttrs, AbsenceWindow window) {
         this.expectAttrs = expectAttrs;
         this.window = window;
         expired = false;
     }
 
     /**
-     * return true if it is certain that expected attributes don't occur during startTime and endTime, else return false
-     * @param appearAttrs
-     * @param occurTime
-     * @return
+     * return true if it is certain that expected attributes don't occur during startTime and endTime, else return false.
      */
-    public void process(List<Object> appearAttrs, long occurTime){
-        if(expired)
+    public void process(List<Object> appearAttrs, long occurTime) {
+        if (expired) {
             throw new IllegalStateException("Expired window can't recieve events");
-        switch(status) {
+        }
+        switch (status) {
             case not_sure:
-                if(occurTime < window.startTime) {
+                if (occurTime < window.startTime) {
                     break;
-                }else if(occurTime >= window.startTime &&
-                        occurTime <= window.endTime) {
-                    if(expectAttrs.equals(appearAttrs)) {
+                } else if (occurTime >= window.startTime
+                    && occurTime <= window.endTime) {
+                    if (expectAttrs.equals(appearAttrs)) {
                         status = OccurStatus.occured;
                     }
                     break;
-                }else{
+                } else {
                     status = OccurStatus.absent;
                     break;
                 }
             case occured:
-                if(occurTime > window.endTime)
+                if (occurTime > window.endTime) {
                     expired = true;
+                }
                 break;
             default:
                 break;
         }
         // reset status
-        if(status == OccurStatus.absent){
+        if (status == OccurStatus.absent) {
             expired = true;
         }
     }
 
-    public OccurStatus checkStatus(){
+    public OccurStatus checkStatus() {
         return status;
     }
-    public boolean checkExpired(){
+
+    public boolean checkExpired() {
         return expired;
     }
-    public AbsenceWindow currWindow(){
+
+    public AbsenceWindow currWindow() {
         return window;
     }
 
-    public String toString(){
+    public String toString() {
         return "expectAttrs=" + expectAttrs + ", status=" + status + ", expired=" + expired + ", window=[" + window + "]";
     }
 }
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/a0fc8708/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/AlertBoltOutputCollectorThreadSafeWrapper.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/AlertBoltOutputCollectorThreadSafeWrapper.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/AlertBoltOutputCollectorThreadSafeWrapper.java
index ca1f622..33d502e 100755
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/AlertBoltOutputCollectorThreadSafeWrapper.java
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/AlertBoltOutputCollectorThreadSafeWrapper.java
@@ -16,37 +16,36 @@
  */
 package org.apache.eagle.alert.engine.evaluator.impl;
 
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.atomic.AtomicLong;
-
 import org.apache.eagle.alert.engine.AlertStreamCollector;
 import org.apache.eagle.alert.engine.model.AlertStreamEvent;
+import backtype.storm.task.OutputCollector;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import backtype.storm.task.OutputCollector;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.atomic.AtomicLong;
 
 /**
- * <h2>Thread Safe Mechanism</h2>
+ * <h2>Thread Safe Mechanism.</h2>
  * <ul>
  * <li>
- *     emit() method is thread-safe enough to be called anywhere asynchronously in multi-thread
+ * emit() method is thread-safe enough to be called anywhere asynchronously in multi-thread
  * </li>
  * <li>
- *     flush() method must be called synchronously, because Storm OutputCollector is not thread-safe
+ * flush() method must be called synchronously, because Storm OutputCollector is not thread-safe
  * </li>
  * </ul>
  */
 public class AlertBoltOutputCollectorThreadSafeWrapper implements AlertStreamCollector {
     private final OutputCollector delegate;
     private final LinkedBlockingQueue<AlertStreamEvent> queue;
-    private final static Logger LOG = LoggerFactory.getLogger(AlertBoltOutputCollectorThreadSafeWrapper.class);
+    private static final Logger LOG = LoggerFactory.getLogger(AlertBoltOutputCollectorThreadSafeWrapper.class);
     private final AtomicLong lastFlushTime = new AtomicLong(System.currentTimeMillis());
     private final AutoAlertFlusher flusher;
-    private final static int MAX_ALERT_DELAY_SECS = 10;
+    private static final int MAX_ALERT_DELAY_SECS = 10;
 
     public AlertBoltOutputCollectorThreadSafeWrapper(OutputCollector outputCollector) {
         this.delegate = outputCollector;
@@ -59,7 +58,7 @@ public class AlertBoltOutputCollectorThreadSafeWrapper implements AlertStreamCol
     private static class AutoAlertFlusher extends Thread {
         private final AlertBoltOutputCollectorThreadSafeWrapper collector;
         private boolean stopped = false;
-        private final static Logger LOG = LoggerFactory.getLogger(AutoAlertFlusher.class);
+        private static final Logger LOG = LoggerFactory.getLogger(AutoAlertFlusher.class);
 
         private AutoAlertFlusher(AlertBoltOutputCollectorThreadSafeWrapper collector) {
             this.collector = collector;
@@ -75,6 +74,7 @@ public class AlertBoltOutputCollectorThreadSafeWrapper implements AlertStreamCol
                 try {
                     Thread.sleep(5000);
                 } catch (InterruptedException ignored) {
+                    // ignored
                 }
             }
             LOG.info("Stopped");
@@ -87,7 +87,8 @@ public class AlertBoltOutputCollectorThreadSafeWrapper implements AlertStreamCol
     }
 
     /**
-     * Emit method can be called in multi-thread
+     * Emit method can be called in multi-thread.
+     *
      * @param event
      */
     @Override
@@ -95,7 +96,7 @@ public class AlertBoltOutputCollectorThreadSafeWrapper implements AlertStreamCol
         try {
             queue.put(event);
         } catch (InterruptedException e) {
-            LOG.error(e.getMessage(),e);
+            LOG.error(e.getMessage(), e);
         }
     }
 
@@ -104,7 +105,7 @@ public class AlertBoltOutputCollectorThreadSafeWrapper implements AlertStreamCol
      */
     @Override
     public void flush() {
-        if(!queue.isEmpty()) {
+        if (!queue.isEmpty()) {
             List<AlertStreamEvent> events = new ArrayList<>();
             queue.drainTo(events);
             events.forEach((event) -> delegate.emit(Arrays.asList(event.getStreamId(), event)));

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/a0fc8708/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/AlertBoltOutputCollectorWrapper.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/AlertBoltOutputCollectorWrapper.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/AlertBoltOutputCollectorWrapper.java
index 8d8a4d2..042fca7 100755
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/AlertBoltOutputCollectorWrapper.java
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/AlertBoltOutputCollectorWrapper.java
@@ -16,20 +16,19 @@
  */
 package org.apache.eagle.alert.engine.evaluator.impl;
 
-import java.util.Arrays;
-
 import org.apache.eagle.alert.engine.AlertStreamCollector;
 import org.apache.eagle.alert.engine.StreamContext;
 import org.apache.eagle.alert.engine.model.AlertStreamEvent;
-
 import backtype.storm.task.OutputCollector;
 
+import java.util.Arrays;
+
 public class AlertBoltOutputCollectorWrapper implements AlertStreamCollector {
     private final OutputCollector delegate;
     private final Object outputLock;
     private final StreamContext streamContext;
 
-    public AlertBoltOutputCollectorWrapper(OutputCollector outputCollector, Object outputLock, StreamContext streamContext){
+    public AlertBoltOutputCollectorWrapper(OutputCollector outputCollector, Object outputLock, StreamContext streamContext) {
         this.delegate = outputCollector;
         this.outputLock = outputLock;
         this.streamContext = streamContext;

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/a0fc8708/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/AlertStreamCallback.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/AlertStreamCallback.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/AlertStreamCallback.java
index b9a109c..ee1853c 100644
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/AlertStreamCallback.java
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/AlertStreamCallback.java
@@ -52,14 +52,12 @@ public class AlertStreamCallback extends StreamCallback {
     }
 
     /**
-     * Possibly more than one event will be triggered for alerting
-     *
-     * @param events
+     * Possibly more than one event will be triggered for alerting.
      */
     @Override
     public void receive(Event[] events) {
         String policyName = context.getPolicyDefinition().getName();
-        CompositePolicyHandler handler = ((PolicyGroupEvaluatorImpl)context.getPolicyEvaluator()).getPolicyHandler(policyName);
+        CompositePolicyHandler handler = ((PolicyGroupEvaluatorImpl) context.getPolicyEvaluator()).getPolicyHandler(policyName);
         if (LOG.isDebugEnabled()) {
             LOG.debug("Generated {} alerts from policy '{}' in {}, index of definiton {} ", events.length, policyName, context.getPolicyEvaluatorId(), currentIndex);
         }
@@ -75,18 +73,21 @@ public class AlertStreamCallback extends StreamCallback {
             event.setCreatedTime(System.currentTimeMillis());
             event.setSchema(definition);
 
-            if (LOG.isDebugEnabled())
+            if (LOG.isDebugEnabled()) {
                 LOG.debug("Generate new alert event: {}", event);
+            }
             try {
                 if (handler == null) {
                     // extreme case: the handler is removed from the evaluator. Just emit.
-                    if (LOG.isDebugEnabled()) LOG.debug(" handler not found when callback received event, directly emit. policy removed? ");
+                    if (LOG.isDebugEnabled()) {
+                        LOG.debug(" handler not found when callback received event, directly emit. policy removed? ");
+                    }
                     collector.emit(event);
                 } else {
                     handler.send(event, currentIndex + 1);
                 }
             } catch (Exception ex) {
-                LOG.error(String.format("send event %s to index %d failed with exception. ",event, currentIndex), ex);
+                LOG.error(String.format("send event %s to index %d failed with exception. ", event, currentIndex), ex);
             }
         }
         context.getPolicyCounter().scope(String.format("%s.%s", this.context.getPolicyDefinition().getName(), "alert_count")).incrBy(events.length);

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/a0fc8708/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/PolicyGroupEvaluatorImpl.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/PolicyGroupEvaluatorImpl.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/PolicyGroupEvaluatorImpl.java
index 26ae19e..eed4b3b 100644
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/PolicyGroupEvaluatorImpl.java
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/PolicyGroupEvaluatorImpl.java
@@ -35,17 +35,17 @@ import java.util.Map;
 public class PolicyGroupEvaluatorImpl implements PolicyGroupEvaluator {
     private static final long serialVersionUID = -5499413193675985288L;
 
-    private final static Logger LOG = LoggerFactory.getLogger(PolicyGroupEvaluatorImpl.class);
+    private static final Logger LOG = LoggerFactory.getLogger(PolicyGroupEvaluatorImpl.class);
 
     private AlertStreamCollector collector;
     // mapping from policy name to PolicyDefinition
-    private volatile Map<String,PolicyDefinition> policyDefinitionMap = new HashMap<>();
+    private volatile Map<String, PolicyDefinition> policyDefinitionMap = new HashMap<>();
     // mapping from policy name to PolicyStreamHandler
-    private volatile Map<String,CompositePolicyHandler> policyStreamHandlerMap = new HashMap<>();
+    private volatile Map<String, CompositePolicyHandler> policyStreamHandlerMap = new HashMap<>();
     private String policyEvaluatorId;
     private StreamContext context;
 
-    public PolicyGroupEvaluatorImpl(String policyEvaluatorId){
+    public PolicyGroupEvaluatorImpl(String policyEvaluatorId) {
         this.policyEvaluatorId = policyEvaluatorId;
     }
 
@@ -67,34 +67,35 @@ public class PolicyGroupEvaluatorImpl implements PolicyGroupEvaluator {
     }
 
     public void close() {
-        for(PolicyStreamHandler handler: policyStreamHandlerMap.values()){
+        for (PolicyStreamHandler handler : policyStreamHandlerMap.values()) {
             try {
                 handler.close();
             } catch (Exception e) {
-                LOG.error("Failed to close handler {}",handler.toString(),e);
+                LOG.error("Failed to close handler {}", handler.toString(), e);
             }
         }
     }
 
     /**
-     * fixme make selection of PolicyStreamHandler to be more efficient
+     * fixme make selection of PolicyStreamHandler to be more efficient.
+     *
      * @param partitionedEvent PartitionedEvent
      */
-    private void dispatch(PartitionedEvent partitionedEvent){
+    private void dispatch(PartitionedEvent partitionedEvent) {
         boolean handled = false;
-        for(Map.Entry<String,CompositePolicyHandler> policyStreamHandler: policyStreamHandlerMap.entrySet()){
-            if(isAcceptedByPolicy(partitionedEvent,policyDefinitionMap.get(policyStreamHandler.getKey()))){
+        for (Map.Entry<String, CompositePolicyHandler> policyStreamHandler : policyStreamHandlerMap.entrySet()) {
+            if (isAcceptedByPolicy(partitionedEvent, policyDefinitionMap.get(policyStreamHandler.getKey()))) {
                 try {
                     handled = true;
                     this.context.counter().scope("eval_count").incr();
                     policyStreamHandler.getValue().send(partitionedEvent.getEvent());
                 } catch (Exception e) {
                     this.context.counter().scope("fail_count").incr();
-                    LOG.error("{} failed to handle {}",policyStreamHandler.getValue(), partitionedEvent.getEvent(), e);
+                    LOG.error("{} failed to handle {}", policyStreamHandler.getValue(), partitionedEvent.getEvent(), e);
                 }
             }
         }
-        if(!handled){
+        if (!handled) {
             this.context.counter().scope("drop_count").incr();
             LOG.warn("Drop stream non-matched event {}, which should not be sent to evaluator", partitionedEvent);
         } else {
@@ -103,22 +104,22 @@ public class PolicyGroupEvaluatorImpl implements PolicyGroupEvaluator {
     }
 
     private static boolean isAcceptedByPolicy(PartitionedEvent event, PolicyDefinition policy) {
-        return policy.getPartitionSpec().contains(event.getPartition()) &&
-                ( policy.getInputStreams().contains(event.getEvent().getStreamId()) ||
-                    policy.getDefinition().getInputStreams().contains(event.getEvent().getStreamId()) );
+        return policy.getPartitionSpec().contains(event.getPartition())
+            && (policy.getInputStreams().contains(event.getEvent().getStreamId())
+            || policy.getDefinition().getInputStreams().contains(event.getEvent().getStreamId()));
     }
 
     @Override
     public void onPolicyChange(List<PolicyDefinition> added, List<PolicyDefinition> removed, List<PolicyDefinition> modified, Map<String, StreamDefinition> sds) {
-        Map<String,PolicyDefinition> copyPolicies = new HashMap<>(policyDefinitionMap);
-        Map<String,CompositePolicyHandler> copyHandlers = new HashMap<>(policyStreamHandlerMap);
-        for(PolicyDefinition pd : added){
+        Map<String, PolicyDefinition> copyPolicies = new HashMap<>(policyDefinitionMap);
+        Map<String, CompositePolicyHandler> copyHandlers = new HashMap<>(policyStreamHandlerMap);
+        for (PolicyDefinition pd : added) {
             inplaceAdd(copyPolicies, copyHandlers, pd, sds);
         }
-        for(PolicyDefinition pd : removed){
+        for (PolicyDefinition pd : removed) {
             inplaceRemove(copyPolicies, copyHandlers, pd);
         }
-        for(PolicyDefinition pd : modified){
+        for (PolicyDefinition pd : modified) {
             inplaceRemove(copyPolicies, copyHandlers, pd);
             inplaceAdd(copyPolicies, copyHandlers, pd, sds);
         }
@@ -132,9 +133,9 @@ public class PolicyGroupEvaluatorImpl implements PolicyGroupEvaluator {
     }
 
     private void inplaceAdd(Map<String, PolicyDefinition> policies, Map<String, CompositePolicyHandler> handlers, PolicyDefinition policy, Map<String, StreamDefinition> sds) {
-        if(handlers.containsKey(policy.getName())){
+        if (handlers.containsKey(policy.getName())) {
             LOG.error("metadata calculation error, try to add existing PolicyDefinition " + policy);
-        }else {
+        } else {
             policies.put(policy.getName(), policy);
             CompositePolicyHandler handler = new CompositePolicyHandler(sds);
             try {
@@ -154,20 +155,20 @@ public class PolicyGroupEvaluatorImpl implements PolicyGroupEvaluator {
         }
     }
 
-    private void inplaceRemove(Map<String, PolicyDefinition> policies, Map<String, CompositePolicyHandler> handlers, PolicyDefinition policy)  {
-        if(handlers.containsKey(policy.getName())) {
+    private void inplaceRemove(Map<String, PolicyDefinition> policies, Map<String, CompositePolicyHandler> handlers, PolicyDefinition policy) {
+        if (handlers.containsKey(policy.getName())) {
             PolicyStreamHandler handler = handlers.get(policy.getName());
             try {
                 handler.close();
             } catch (Exception e) {
-                LOG.error("Failed to close handler {}",handler,e);
-            }finally {
+                LOG.error("Failed to close handler {}", handler, e);
+            } finally {
                 policies.remove(policy.getName());
                 handlers.remove(policy.getName());
-                LOG.info("Removed policy: {}",policy);
+                LOG.info("Removed policy: {}", policy);
             }
         } else {
-            LOG.error("metadata calculation error, try to remove nonexisting PolicyDefinition: "+policy);
+            LOG.error("metadata calculation error, try to remove nonexisting PolicyDefinition: " + policy);
         }
     }
 

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/a0fc8708/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/SiddhiDefinitionAdapter.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/SiddhiDefinitionAdapter.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/SiddhiDefinitionAdapter.java
index 5b83abb..9b9fcac 100644
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/SiddhiDefinitionAdapter.java
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/SiddhiDefinitionAdapter.java
@@ -16,119 +16,118 @@
  */
 package org.apache.eagle.alert.engine.evaluator.impl;
 
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-import org.apache.commons.lang3.StringUtils;
 import org.apache.eagle.alert.engine.coordinator.StreamColumn;
 import org.apache.eagle.alert.engine.coordinator.StreamDefinition;
+import com.google.common.base.Preconditions;
+import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.wso2.siddhi.query.api.definition.AbstractDefinition;
 import org.wso2.siddhi.query.api.definition.Attribute;
 
-import com.google.common.base.Preconditions;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
 
 public class SiddhiDefinitionAdapter {
-    private final static Logger LOG = LoggerFactory.getLogger(SiddhiDefinitionAdapter.class);
-    public final static String DEFINE_STREAM_TEMPLATE =  "define stream %s ( %s );";
+    private static final Logger LOG = LoggerFactory.getLogger(SiddhiDefinitionAdapter.class);
+    public static final String DEFINE_STREAM_TEMPLATE = "define stream %s ( %s );";
 
-    public static String buildStreamDefinition(StreamDefinition streamDefinition){
+    public static String buildStreamDefinition(StreamDefinition streamDefinition) {
         List<String> columns = new ArrayList<>();
-        Preconditions.checkNotNull(streamDefinition,"StreamDefinition is null");
-        if(streamDefinition.getColumns()!=null) {
+        Preconditions.checkNotNull(streamDefinition, "StreamDefinition is null");
+        if (streamDefinition.getColumns() != null) {
             for (StreamColumn column : streamDefinition.getColumns()) {
                 columns.add(String.format("%s %s", column.getName(), convertToSiddhiAttributeType(column.getType()).toString().toLowerCase()));
             }
-        }else{
-            LOG.warn("No columns found for stream {}"+streamDefinition.getStreamId());
+        } else {
+            LOG.warn("No columns found for stream {}" + streamDefinition.getStreamId());
         }
-        return String.format(DEFINE_STREAM_TEMPLATE, streamDefinition.getStreamId(),StringUtils.join(columns,","));
+        return String.format(DEFINE_STREAM_TEMPLATE, streamDefinition.getStreamId(), StringUtils.join(columns, ","));
     }
 
-    public static Attribute.Type convertToSiddhiAttributeType(StreamColumn.Type type){
-        if(_EAGLE_SIDDHI_TYPE_MAPPING.containsKey(type)){
+    public static Attribute.Type convertToSiddhiAttributeType(StreamColumn.Type type) {
+        if (_EAGLE_SIDDHI_TYPE_MAPPING.containsKey(type)) {
             return _EAGLE_SIDDHI_TYPE_MAPPING.get(type);
         }
 
-        throw new IllegalArgumentException("Unknown stream type: "+type);
+        throw new IllegalArgumentException("Unknown stream type: " + type);
     }
 
-    public static Class<?> convertToJavaAttributeType(StreamColumn.Type type){
-        if(_EAGLE_JAVA_TYPE_MAPPING.containsKey(type)){
+    public static Class<?> convertToJavaAttributeType(StreamColumn.Type type) {
+        if (_EAGLE_JAVA_TYPE_MAPPING.containsKey(type)) {
             return _EAGLE_JAVA_TYPE_MAPPING.get(type);
         }
 
-        throw new IllegalArgumentException("Unknown stream type: "+type);
+        throw new IllegalArgumentException("Unknown stream type: " + type);
     }
 
-    public static StreamColumn.Type convertFromJavaAttributeType(Class<?> type){
-        if(_JAVA_EAGLE_TYPE_MAPPING.containsKey(type)){
+    public static StreamColumn.Type convertFromJavaAttributeType(Class<?> type) {
+        if (_JAVA_EAGLE_TYPE_MAPPING.containsKey(type)) {
             return _JAVA_EAGLE_TYPE_MAPPING.get(type);
         }
 
-        throw new IllegalArgumentException("Unknown stream type: "+type);
+        throw new IllegalArgumentException("Unknown stream type: " + type);
     }
 
-    public static StreamColumn.Type convertFromSiddhiAttributeType(Attribute.Type type){
-        if(_SIDDHI_EAGLE_TYPE_MAPPING.containsKey(type)){
+    public static StreamColumn.Type convertFromSiddhiAttributeType(Attribute.Type type) {
+        if (_SIDDHI_EAGLE_TYPE_MAPPING.containsKey(type)) {
             return _SIDDHI_EAGLE_TYPE_MAPPING.get(type);
         }
 
-        throw new IllegalArgumentException("Unknown siddhi type: "+type);
+        throw new IllegalArgumentException("Unknown siddhi type: " + type);
     }
 
     /**
      * public enum Type {
-     *   STRING, INT, LONG, FLOAT, DOUBLE, BOOL, OBJECT
-     * }
+     * STRING, INT, LONG, FLOAT, DOUBLE, BOOL, OBJECT
+     * }.
      */
-    private final static Map<StreamColumn.Type,Attribute.Type> _EAGLE_SIDDHI_TYPE_MAPPING = new HashMap<>();
-    private final static Map<StreamColumn.Type,Class<?>> _EAGLE_JAVA_TYPE_MAPPING = new HashMap<>();
-    private final static Map<Class<?>,StreamColumn.Type> _JAVA_EAGLE_TYPE_MAPPING = new HashMap<>();
-    private final static Map<Attribute.Type,StreamColumn.Type> _SIDDHI_EAGLE_TYPE_MAPPING = new HashMap<>();
+    private static final Map<StreamColumn.Type, Attribute.Type> _EAGLE_SIDDHI_TYPE_MAPPING = new HashMap<>();
+    private static final Map<StreamColumn.Type, Class<?>> _EAGLE_JAVA_TYPE_MAPPING = new HashMap<>();
+    private static final Map<Class<?>, StreamColumn.Type> _JAVA_EAGLE_TYPE_MAPPING = new HashMap<>();
+    private static final Map<Attribute.Type, StreamColumn.Type> _SIDDHI_EAGLE_TYPE_MAPPING = new HashMap<>();
 
     static {
-        _EAGLE_SIDDHI_TYPE_MAPPING.put(StreamColumn.Type.STRING,Attribute.Type.STRING);
-        _EAGLE_SIDDHI_TYPE_MAPPING.put(StreamColumn.Type.INT,Attribute.Type.INT);
-        _EAGLE_SIDDHI_TYPE_MAPPING.put(StreamColumn.Type.LONG,Attribute.Type.LONG);
-        _EAGLE_SIDDHI_TYPE_MAPPING.put(StreamColumn.Type.FLOAT,Attribute.Type.FLOAT);
-        _EAGLE_SIDDHI_TYPE_MAPPING.put(StreamColumn.Type.DOUBLE,Attribute.Type.DOUBLE);
-        _EAGLE_SIDDHI_TYPE_MAPPING.put(StreamColumn.Type.BOOL,Attribute.Type.BOOL);
-        _EAGLE_SIDDHI_TYPE_MAPPING.put(StreamColumn.Type.OBJECT,Attribute.Type.OBJECT);
-
-        _EAGLE_JAVA_TYPE_MAPPING.put(StreamColumn.Type.STRING,String.class);
-        _EAGLE_JAVA_TYPE_MAPPING.put(StreamColumn.Type.INT,Integer.class);
-        _EAGLE_JAVA_TYPE_MAPPING.put(StreamColumn.Type.LONG,Long.class);
-        _EAGLE_JAVA_TYPE_MAPPING.put(StreamColumn.Type.FLOAT,Float.class);
-        _EAGLE_JAVA_TYPE_MAPPING.put(StreamColumn.Type.DOUBLE,Double.class);
-        _EAGLE_JAVA_TYPE_MAPPING.put(StreamColumn.Type.BOOL,Boolean.class);
-        _EAGLE_JAVA_TYPE_MAPPING.put(StreamColumn.Type.OBJECT,Object.class);
-
-        _JAVA_EAGLE_TYPE_MAPPING.put(String.class,StreamColumn.Type.STRING);
-        _JAVA_EAGLE_TYPE_MAPPING.put(Integer.class,StreamColumn.Type.INT);
-        _JAVA_EAGLE_TYPE_MAPPING.put(Long.class,StreamColumn.Type.LONG);
-        _JAVA_EAGLE_TYPE_MAPPING.put(Float.class,StreamColumn.Type.FLOAT);
-        _JAVA_EAGLE_TYPE_MAPPING.put(Double.class,StreamColumn.Type.DOUBLE);
-        _JAVA_EAGLE_TYPE_MAPPING.put(Boolean.class,StreamColumn.Type.BOOL);
-        _JAVA_EAGLE_TYPE_MAPPING.put(Object.class,StreamColumn.Type.OBJECT);
-
-        _SIDDHI_EAGLE_TYPE_MAPPING.put(Attribute.Type.STRING,StreamColumn.Type.STRING);
-        _SIDDHI_EAGLE_TYPE_MAPPING.put(Attribute.Type.INT,StreamColumn.Type.INT);
-        _SIDDHI_EAGLE_TYPE_MAPPING.put(Attribute.Type.LONG,StreamColumn.Type.LONG);
-        _SIDDHI_EAGLE_TYPE_MAPPING.put(Attribute.Type.FLOAT,StreamColumn.Type.FLOAT);
-        _SIDDHI_EAGLE_TYPE_MAPPING.put(Attribute.Type.DOUBLE,StreamColumn.Type.DOUBLE);
-        _SIDDHI_EAGLE_TYPE_MAPPING.put(Attribute.Type.BOOL,StreamColumn.Type.BOOL);
-        _SIDDHI_EAGLE_TYPE_MAPPING.put(Attribute.Type.OBJECT,StreamColumn.Type.OBJECT);
+        _EAGLE_SIDDHI_TYPE_MAPPING.put(StreamColumn.Type.STRING, Attribute.Type.STRING);
+        _EAGLE_SIDDHI_TYPE_MAPPING.put(StreamColumn.Type.INT, Attribute.Type.INT);
+        _EAGLE_SIDDHI_TYPE_MAPPING.put(StreamColumn.Type.LONG, Attribute.Type.LONG);
+        _EAGLE_SIDDHI_TYPE_MAPPING.put(StreamColumn.Type.FLOAT, Attribute.Type.FLOAT);
+        _EAGLE_SIDDHI_TYPE_MAPPING.put(StreamColumn.Type.DOUBLE, Attribute.Type.DOUBLE);
+        _EAGLE_SIDDHI_TYPE_MAPPING.put(StreamColumn.Type.BOOL, Attribute.Type.BOOL);
+        _EAGLE_SIDDHI_TYPE_MAPPING.put(StreamColumn.Type.OBJECT, Attribute.Type.OBJECT);
+
+        _EAGLE_JAVA_TYPE_MAPPING.put(StreamColumn.Type.STRING, String.class);
+        _EAGLE_JAVA_TYPE_MAPPING.put(StreamColumn.Type.INT, Integer.class);
+        _EAGLE_JAVA_TYPE_MAPPING.put(StreamColumn.Type.LONG, Long.class);
+        _EAGLE_JAVA_TYPE_MAPPING.put(StreamColumn.Type.FLOAT, Float.class);
+        _EAGLE_JAVA_TYPE_MAPPING.put(StreamColumn.Type.DOUBLE, Double.class);
+        _EAGLE_JAVA_TYPE_MAPPING.put(StreamColumn.Type.BOOL, Boolean.class);
+        _EAGLE_JAVA_TYPE_MAPPING.put(StreamColumn.Type.OBJECT, Object.class);
+
+        _JAVA_EAGLE_TYPE_MAPPING.put(String.class, StreamColumn.Type.STRING);
+        _JAVA_EAGLE_TYPE_MAPPING.put(Integer.class, StreamColumn.Type.INT);
+        _JAVA_EAGLE_TYPE_MAPPING.put(Long.class, StreamColumn.Type.LONG);
+        _JAVA_EAGLE_TYPE_MAPPING.put(Float.class, StreamColumn.Type.FLOAT);
+        _JAVA_EAGLE_TYPE_MAPPING.put(Double.class, StreamColumn.Type.DOUBLE);
+        _JAVA_EAGLE_TYPE_MAPPING.put(Boolean.class, StreamColumn.Type.BOOL);
+        _JAVA_EAGLE_TYPE_MAPPING.put(Object.class, StreamColumn.Type.OBJECT);
+
+        _SIDDHI_EAGLE_TYPE_MAPPING.put(Attribute.Type.STRING, StreamColumn.Type.STRING);
+        _SIDDHI_EAGLE_TYPE_MAPPING.put(Attribute.Type.INT, StreamColumn.Type.INT);
+        _SIDDHI_EAGLE_TYPE_MAPPING.put(Attribute.Type.LONG, StreamColumn.Type.LONG);
+        _SIDDHI_EAGLE_TYPE_MAPPING.put(Attribute.Type.FLOAT, StreamColumn.Type.FLOAT);
+        _SIDDHI_EAGLE_TYPE_MAPPING.put(Attribute.Type.DOUBLE, StreamColumn.Type.DOUBLE);
+        _SIDDHI_EAGLE_TYPE_MAPPING.put(Attribute.Type.BOOL, StreamColumn.Type.BOOL);
+        _SIDDHI_EAGLE_TYPE_MAPPING.put(Attribute.Type.OBJECT, StreamColumn.Type.OBJECT);
     }
 
-    public static StreamDefinition convertFromSiddiDefinition(AbstractDefinition siddhiDefinition){
+    public static StreamDefinition convertFromSiddiDefinition(AbstractDefinition siddhiDefinition) {
         StreamDefinition streamDefinition = new StreamDefinition();
         streamDefinition.setStreamId(siddhiDefinition.getId());
         List<StreamColumn> columns = new ArrayList<>(siddhiDefinition.getAttributeNameArray().length);
-        for(Attribute attribute:siddhiDefinition.getAttributeList()){
+        for (Attribute attribute : siddhiDefinition.getAttributeList()) {
             StreamColumn column = new StreamColumn();
             column.setType(convertFromSiddhiAttributeType(attribute.getType()));
             column.setName(attribute.getName());
@@ -136,7 +135,7 @@ public class SiddhiDefinitionAdapter {
         }
         streamDefinition.setColumns(columns);
         streamDefinition.setTimeseries(true);
-        streamDefinition.setDescription("Auto-generated stream schema from siddhi for "+siddhiDefinition.getId());
+        streamDefinition.setDescription("Auto-generated stream schema from siddhi for " + siddhiDefinition.getId());
         return streamDefinition;
     }
 }
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/a0fc8708/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/SiddhiPolicyHandler.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/SiddhiPolicyHandler.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/SiddhiPolicyHandler.java
index 481e3af..e7ed56f 100755
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/SiddhiPolicyHandler.java
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/SiddhiPolicyHandler.java
@@ -34,7 +34,7 @@ import java.util.List;
 import java.util.Map;
 
 public class SiddhiPolicyHandler implements PolicyStreamHandler {
-    private final static Logger LOG = LoggerFactory.getLogger(SiddhiPolicyHandler.class);
+    private static final Logger LOG = LoggerFactory.getLogger(SiddhiPolicyHandler.class);
     private ExecutionPlanRuntime executionRuntime;
     private SiddhiManager siddhiManager;
     private Map<String, StreamDefinition> sds;
@@ -43,7 +43,7 @@ public class SiddhiPolicyHandler implements PolicyStreamHandler {
 
     private int currentIndex = 0; // the index of current definition statement inside the policy definition
 
-    public SiddhiPolicyHandler(Map<String, StreamDefinition> sds, int index){
+    public SiddhiPolicyHandler(Map<String, StreamDefinition> sds, int index) {
         this.sds = sds;
         this.currentIndex = index;
     }
@@ -59,44 +59,46 @@ public class SiddhiPolicyHandler implements PolicyStreamHandler {
             coreDefinition.setOutputStreams(policyDefinition.getOutputStreams());
         }
 
-        for(String inputStream : coreDefinition.getInputStreams()) {
+        for (String inputStream : coreDefinition.getInputStreams()) {
             builder.append(SiddhiDefinitionAdapter.buildStreamDefinition(sds.get(inputStream)));
             builder.append("\n");
         }
         builder.append(coreDefinition.value);
-        if(LOG.isDebugEnabled()) LOG.debug("Generated siddhi execution plan: {} from definition: {}", builder.toString(), coreDefinition);
+        if (LOG.isDebugEnabled()) {
+            LOG.debug("Generated siddhi execution plan: {} from definition: {}", builder.toString(), coreDefinition);
+        }
         return builder.toString();
     }
 
     @Override
     public void prepare(final Collector<AlertStreamEvent> collector, PolicyHandlerContext context) throws Exception {
-        LOG.info("Initializing handler for policy {}",context.getPolicyDefinition());
+        LOG.info("Initializing handler for policy {}", context.getPolicyDefinition());
         this.policy = context.getPolicyDefinition();
         this.siddhiManager = new SiddhiManager();
         String plan = generateExecutionPlan(policy, sds);
         try {
             this.executionRuntime = siddhiManager.createExecutionPlanRuntime(plan);
-            LOG.info("Created siddhi runtime {}",executionRuntime.getName());
-        }catch (Exception parserException){
-            LOG.error("Failed to create siddhi runtime for policy: {}, siddhi plan: \n\n{}\n",context.getPolicyDefinition().getName(),plan,parserException);
+            LOG.info("Created siddhi runtime {}", executionRuntime.getName());
+        } catch (Exception parserException) {
+            LOG.error("Failed to create siddhi runtime for policy: {}, siddhi plan: \n\n{}\n", context.getPolicyDefinition().getName(), plan, parserException);
             throw parserException;
         }
 
         // add output stream callback
         List<String> outputStreams = getOutputStreams(policy);
-        for(final String outputStream: outputStreams) {
+        for (final String outputStream : outputStreams) {
             if (executionRuntime.getStreamDefinitionMap().containsKey(outputStream)) {
                 this.executionRuntime.addCallback(outputStream,
-                        new AlertStreamCallback(
-                                outputStream, SiddhiDefinitionAdapter.convertFromSiddiDefinition(executionRuntime.getStreamDefinitionMap().get(outputStream))
-                                , collector, context, currentIndex));
+                    new AlertStreamCallback(
+                        outputStream, SiddhiDefinitionAdapter.convertFromSiddiDefinition(executionRuntime.getStreamDefinitionMap().get(outputStream)),
+                        collector, context, currentIndex));
             } else {
                 throw new IllegalStateException("Undefined output stream " + outputStream);
             }
         }
         this.executionRuntime.start();
         this.context = context;
-        LOG.info("Initialized policy handler for policy: {}",policy.getName());
+        LOG.info("Initialized policy handler for policy: {}", policy.getName());
     }
 
     protected List<String> getOutputStreams(PolicyDefinition policy) {
@@ -104,29 +106,29 @@ public class SiddhiPolicyHandler implements PolicyStreamHandler {
     }
 
     public void send(StreamEvent event) throws Exception {
-        context.getPolicyCounter().scope(String.format("%s.%s",this.context.getPolicyDefinition().getName(),"receive_count")).incr();
+        context.getPolicyCounter().scope(String.format("%s.%s", this.context.getPolicyDefinition().getName(), "receive_count")).incr();
         String streamId = event.getStreamId();
         InputHandler inputHandler = executionRuntime.getInputHandler(streamId);
-        if(inputHandler != null){
-            context.getPolicyCounter().scope(String.format("%s.%s",this.context.getPolicyDefinition().getName(),"eval_count")).incr();
-            inputHandler.send(event.getTimestamp(),event.getData());
-            
+        if (inputHandler != null) {
+            context.getPolicyCounter().scope(String.format("%s.%s", this.context.getPolicyDefinition().getName(), "eval_count")).incr();
+            inputHandler.send(event.getTimestamp(), event.getData());
+
             if (LOG.isDebugEnabled()) {
                 LOG.debug("sent event to siddhi stream {} ", streamId);
             }
-        }else{
-            context.getPolicyCounter().scope(String.format("%s.%s",this.context.getPolicyDefinition().getName(),"drop_count")).incr();
-            LOG.warn("No input handler found for stream {}",streamId);
+        } else {
+            context.getPolicyCounter().scope(String.format("%s.%s", this.context.getPolicyDefinition().getName(), "drop_count")).incr();
+            LOG.warn("No input handler found for stream {}", streamId);
         }
     }
 
     public void close() throws Exception {
-        LOG.info("Closing handler for policy {}",this.policy.getName());
+        LOG.info("Closing handler for policy {}", this.policy.getName());
         this.executionRuntime.shutdown();
-        LOG.info("Shutdown siddhi runtime {}",this.executionRuntime.getName());
+        LOG.info("Shutdown siddhi runtime {}", this.executionRuntime.getName());
         this.siddhiManager.shutdown();
-        LOG.info("Shutdown siddhi manager {}",this.siddhiManager);
-        LOG.info("Closed handler for policy {}",this.policy.getName());
+        LOG.info("Shutdown siddhi manager {}", this.siddhiManager);
+        LOG.info("Closed handler for policy {}", this.policy.getName());
     }
 
     @Override

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/a0fc8708/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/SiddhiPolicyStateHandler.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/SiddhiPolicyStateHandler.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/SiddhiPolicyStateHandler.java
index 43b8f30..11f484d 100644
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/SiddhiPolicyStateHandler.java
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/SiddhiPolicyStateHandler.java
@@ -30,7 +30,7 @@ import java.util.Map;
  */
 public class SiddhiPolicyStateHandler extends SiddhiPolicyHandler {
 
-    private final static Logger LOG = LoggerFactory.getLogger(SiddhiPolicyStateHandler.class);
+    private static final Logger LOG = LoggerFactory.getLogger(SiddhiPolicyStateHandler.class);
 
     public SiddhiPolicyStateHandler(Map<String, StreamDefinition> sds, int index) {
         super(sds, index);
@@ -40,12 +40,14 @@ public class SiddhiPolicyStateHandler extends SiddhiPolicyHandler {
     protected String generateExecutionPlan(PolicyDefinition policyDefinition, Map<String, StreamDefinition> sds) throws StreamDefinitionNotFoundException {
         StringBuilder builder = new StringBuilder();
         PolicyDefinition.Definition stateDefiniton = policyDefinition.getStateDefinition();
-        for(String inputStream : stateDefiniton.getInputStreams()) { // the state stream follow the output stream of the policy definition
+        for (String inputStream : stateDefiniton.getInputStreams()) { // the state stream follow the output stream of the policy definition
             builder.append(SiddhiDefinitionAdapter.buildStreamDefinition(sds.get(inputStream)));
             builder.append("\n");
         }
         builder.append(stateDefiniton.value);
-        if(LOG.isDebugEnabled()) LOG.debug("Generated siddhi state execution plan: {} from definiton: {}", builder.toString(), stateDefiniton);
+        if (LOG.isDebugEnabled()) {
+            LOG.debug("Generated siddhi state execution plan: {} from definiton: {}", builder.toString(), stateDefiniton);
+        }
         return builder.toString();
     }
 

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/a0fc8708/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/nodata/DistinctValuesInTimeBatchWindow.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/nodata/DistinctValuesInTimeBatchWindow.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/nodata/DistinctValuesInTimeBatchWindow.java
index 357504e..ef806fb 100644
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/nodata/DistinctValuesInTimeBatchWindow.java
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/nodata/DistinctValuesInTimeBatchWindow.java
@@ -16,6 +16,10 @@
  */
 package org.apache.eagle.alert.engine.evaluator.nodata;
 
+import org.apache.eagle.alert.engine.model.StreamEvent;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
@@ -24,105 +28,101 @@ import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 
-import org.apache.eagle.alert.engine.model.StreamEvent;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 public class DistinctValuesInTimeBatchWindow {
 
-	private static final Logger LOG = LoggerFactory.getLogger(DistinctValuesInTimeBatchWindow.class);
-	
-	private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
-	
-	// wisb (what is should be) set for expected full set value of multiple columns
-	@SuppressWarnings("rawtypes")
-	private volatile Set wisb = new HashSet();
-	
-	private NoDataPolicyTimeBatchHandler handler;
-	
-	/**
-	 * map from value to max timestamp for this value
-	 */
-	private Map<Object, Long> valueMaxTimeMap = new HashMap<>();
-	
-	private long startTime = -1;
-	private long nextEmitTime = -1;
-	private long timeInMilliSeconds;
-
-	public DistinctValuesInTimeBatchWindow(NoDataPolicyTimeBatchHandler handler, 
-			long timeInMilliSeconds, @SuppressWarnings("rawtypes") Set wisb) {
-		this.handler = handler;
-		this.timeInMilliSeconds = timeInMilliSeconds;
-		if (wisb != null) {
-			this.wisb = wisb;
-		}
-	}
-
-	public Map<Object, Long> distinctValues() {
-		return valueMaxTimeMap;
-	}
-	
-	public void send(StreamEvent event, Object value, long timestamp) {
-		synchronized(this) {
-			if (startTime < 0) {
-				startTime = System.currentTimeMillis();
-				
-				scheduler.scheduleAtFixedRate(new Runnable() {
-
-					@SuppressWarnings({ "unchecked", "rawtypes" })
-					@Override
-					public void run() {
-						try {
-							LOG.info("{}/{}: {}", startTime, nextEmitTime, valueMaxTimeMap.keySet());
-							synchronized (valueMaxTimeMap) {
-								boolean sendAlerts = false;
-								
-								if (nextEmitTime < 0) {
-									nextEmitTime = startTime + timeInMilliSeconds;
-								}
-								
-								if (System.currentTimeMillis() > nextEmitTime) {
-									startTime = nextEmitTime;
-									nextEmitTime += timeInMilliSeconds;
-									sendAlerts = true;
-								} else {
-									sendAlerts = false;
-								}
-								
-								if (sendAlerts) {
-									// alert
-									handler.compareAndEmit(wisb, distinctValues().keySet(), event);
-									LOG.info("alert for wiri: {} compares to wisb: {}", distinctValues().keySet(), wisb);
-									
-									if (distinctValues().keySet().size() > 0) {
-										wisb = new HashSet(distinctValues().keySet());
-									}
-									valueMaxTimeMap.clear();
-									LOG.info("Clear wiri & update wisb to {}", wisb);
-								}
-							}
-						} catch (Throwable t) {
-							LOG.error("failed to run batch window for gap alert", t);
-						}
-					}
-					
-				}, 0, timeInMilliSeconds / 2, TimeUnit.MILLISECONDS);
-			}
-		}
-		
-		if (valueMaxTimeMap.containsKey(value)) {
-			// remove that entry with old timestamp in timeSortedMap
-			long oldTime = valueMaxTimeMap.get(value);
-			if (oldTime >= timestamp) {
-				// no any effect as the new timestamp is equal or even less than
-				// old timestamp
-				return;
-			}
-		}
-		// update new timestamp in valueMaxTimeMap
-		valueMaxTimeMap.put(value, timestamp);
-		
-		LOG.info("sent: {} with start: {}/next: {}", valueMaxTimeMap.keySet(), startTime, nextEmitTime);
-	}
+    private static final Logger LOG = LoggerFactory.getLogger(DistinctValuesInTimeBatchWindow.class);
+
+    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
+
+    // wisb (what is should be) set for expected full set value of multiple columns
+    @SuppressWarnings("rawtypes")
+    private volatile Set wisb = new HashSet();
+
+    private NoDataPolicyTimeBatchHandler handler;
+
+    /**
+     * map from value to max timestamp for this value.
+     */
+    private Map<Object, Long> valueMaxTimeMap = new HashMap<>();
+
+    private long startTime = -1;
+    private long nextEmitTime = -1;
+    private long timeInMilliSeconds;
+
+    public DistinctValuesInTimeBatchWindow(NoDataPolicyTimeBatchHandler handler,
+                                           long timeInMilliSeconds, @SuppressWarnings("rawtypes") Set wisb) {
+        this.handler = handler;
+        this.timeInMilliSeconds = timeInMilliSeconds;
+        if (wisb != null) {
+            this.wisb = wisb;
+        }
+    }
+
+    public Map<Object, Long> distinctValues() {
+        return valueMaxTimeMap;
+    }
+
+    public void send(StreamEvent event, Object value, long timestamp) {
+        synchronized (this) {
+            if (startTime < 0) {
+                startTime = System.currentTimeMillis();
+
+                scheduler.scheduleAtFixedRate(new Runnable() {
+
+                    @SuppressWarnings( {"unchecked", "rawtypes"})
+                    @Override
+                    public void run() {
+                        try {
+                            LOG.info("{}/{}: {}", startTime, nextEmitTime, valueMaxTimeMap.keySet());
+                            synchronized (valueMaxTimeMap) {
+                                boolean sendAlerts = false;
+
+                                if (nextEmitTime < 0) {
+                                    nextEmitTime = startTime + timeInMilliSeconds;
+                                }
+
+                                if (System.currentTimeMillis() > nextEmitTime) {
+                                    startTime = nextEmitTime;
+                                    nextEmitTime += timeInMilliSeconds;
+                                    sendAlerts = true;
+                                } else {
+                                    sendAlerts = false;
+                                }
+
+                                if (sendAlerts) {
+                                    // alert
+                                    handler.compareAndEmit(wisb, distinctValues().keySet(), event);
+                                    LOG.info("alert for wiri: {} compares to wisb: {}", distinctValues().keySet(), wisb);
+
+                                    if (distinctValues().keySet().size() > 0) {
+                                        wisb = new HashSet(distinctValues().keySet());
+                                    }
+                                    valueMaxTimeMap.clear();
+                                    LOG.info("Clear wiri & update wisb to {}", wisb);
+                                }
+                            }
+                        } catch (Throwable t) {
+                            LOG.error("failed to run batch window for gap alert", t);
+                        }
+                    }
+
+                }, 0, timeInMilliSeconds / 2, TimeUnit.MILLISECONDS);
+            }
+        }
+
+        if (valueMaxTimeMap.containsKey(value)) {
+            // remove that entry with old timestamp in timeSortedMap
+            long oldTime = valueMaxTimeMap.get(value);
+            if (oldTime >= timestamp) {
+                // no any effect as the new timestamp is equal or even less than
+                // old timestamp
+                return;
+            }
+        }
+        // update new timestamp in valueMaxTimeMap
+        valueMaxTimeMap.put(value, timestamp);
+
+        LOG.info("sent: {} with start: {}/next: {}", valueMaxTimeMap.keySet(), startTime, nextEmitTime);
+    }
 
 }

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/a0fc8708/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/nodata/DistinctValuesInTimeWindow.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/nodata/DistinctValuesInTimeWindow.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/nodata/DistinctValuesInTimeWindow.java
index 676357a..4aae040 100644
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/nodata/DistinctValuesInTimeWindow.java
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/nodata/DistinctValuesInTimeWindow.java
@@ -16,70 +16,69 @@
  */
 package org.apache.eagle.alert.engine.evaluator.nodata;
 
-import java.util.Comparator;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.Map;
-import java.util.SortedMap;
-import java.util.TreeMap;
-
 import org.apache.commons.lang.builder.HashCodeBuilder;
 
+import java.util.*;
+
 /**
  * Since 6/28/16.
  * to get distinct values within a specified time window
  * valueMaxTimeMap : each distinct value is associated with max timestamp it ever had
  * timeSortedMap : map sorted by timestamp first and then value
- * With the above 2 data structure, we can get distinct values in LOG(N)
+ * With the above 2 data structure, we can get distinct values in LOG(N).
  */
 public class DistinctValuesInTimeWindow {
-    public static class ValueAndTime{
+    public static class ValueAndTime {
         Object value;
         long timestamp;
-        public ValueAndTime(Object value, long timestamp){
+
+        public ValueAndTime(Object value, long timestamp) {
             this.value = value;
             this.timestamp = timestamp;
         }
 
-        public String toString(){
+        public String toString() {
             return "[" + value + "," + timestamp + "]";
         }
 
-        public int hashCode(){
+        public int hashCode() {
             return new HashCodeBuilder().append(value).append(timestamp).toHashCode();
         }
 
-        public boolean equals(Object that){
-            if(!(that instanceof ValueAndTime))
+        public boolean equals(Object that) {
+            if (!(that instanceof ValueAndTime)) {
                 return false;
-            ValueAndTime another = (ValueAndTime)that;
+            }
+            ValueAndTime another = (ValueAndTime) that;
             return another.timestamp == this.timestamp && another.value.equals(this.value);
         }
     }
 
-    public static class ValueAndTimeComparator implements Comparator<ValueAndTime>{
+    public static class ValueAndTimeComparator implements Comparator<ValueAndTime> {
         @Override
         public int compare(ValueAndTime o1, ValueAndTime o2) {
-            if(o1.timestamp != o2.timestamp)
+            if (o1.timestamp != o2.timestamp) {
                 return (o1.timestamp > o2.timestamp) ? 1 : -1;
-            if(o1.value.equals(o2.value))
+            }
+            if (o1.value.equals(o2.value)) {
                 return 0;
-            else {
+            } else {
                 // this is not strictly correct, but I don't want to write too many comparators here :-)
-                if(o1.hashCode() > o2.hashCode())
+                if (o1.hashCode() > o2.hashCode()) {
                     return 1;
-                else
+                } else {
                     return -1;
+                }
             }
         }
     }
 
     /**
-     * map from value to max timestamp for this value
+     * map from value to max timestamp for this value.
      */
     private Map<Object, Long> valueMaxTimeMap = new HashMap<>();
     /**
-     * map sorted by time(max timestamp for the value) and then value
+     * map sorted by time(max timestamp for the value) and then value.
      */
     private SortedMap<ValueAndTime, ValueAndTime> timeSortedMap = new TreeMap<>(new ValueAndTimeComparator());
     private long maxTimestamp = 0L;
@@ -87,20 +86,20 @@ public class DistinctValuesInTimeWindow {
     private boolean windowSlided;
 
     /**
-     * @param window - milliseconds
+     * @param window - milliseconds.
      */
-    public DistinctValuesInTimeWindow(long window){
+    public DistinctValuesInTimeWindow(long window) {
         this.window = window;
     }
 
-    public void send(Object value, long timestamp){
+    public void send(Object value, long timestamp) {
         ValueAndTime vt = new ValueAndTime(value, timestamp);
 
         // todo think of time out of order
-        if(valueMaxTimeMap.containsKey(value)){
+        if (valueMaxTimeMap.containsKey(value)) {
             // remove that entry with old timestamp in timeSortedMap
             long oldTime = valueMaxTimeMap.get(value);
-            if(oldTime >= timestamp){
+            if (oldTime >= timestamp) {
                 // no any effect as the new timestamp is equal or even less than old timestamp
                 return;
             }
@@ -117,25 +116,25 @@ public class DistinctValuesInTimeWindow {
 
         // check if some values should be evicted because of time window
         Iterator<Map.Entry<ValueAndTime, ValueAndTime>> it = timeSortedMap.entrySet().iterator();
-        while(it.hasNext()){
+        while (it.hasNext()) {
             Map.Entry<ValueAndTime, ValueAndTime> entry = it.next();
-            if(entry.getKey().timestamp < maxTimestamp - window){
+            if (entry.getKey().timestamp < maxTimestamp - window) {
                 // should remove the entry in valueMaxTimeMap and timeSortedMap
                 valueMaxTimeMap.remove(entry.getKey().value);
                 windowSlided = true;
 
                 it.remove();
-            }else {
+            } else {
                 break;
             }
         }
     }
 
-    public Map<Object, Long> distinctValues(){
+    public Map<Object, Long> distinctValues() {
         return valueMaxTimeMap;
     }
 
-    public boolean windowSlided(){
+    public boolean windowSlided() {
         return windowSlided;
     }
 }

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/a0fc8708/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/nodata/NoDataPolicyHandler.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/nodata/NoDataPolicyHandler.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/nodata/NoDataPolicyHandler.java
index 8f37b93..ec6e6e9 100644
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/nodata/NoDataPolicyHandler.java
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/nodata/NoDataPolicyHandler.java
@@ -16,7 +16,6 @@
  */
 package org.apache.eagle.alert.engine.evaluator.nodata;
 
-import org.apache.commons.collections.CollectionUtils;
 import org.apache.eagle.alert.engine.Collector;
 import org.apache.eagle.alert.engine.coordinator.PolicyDefinition;
 import org.apache.eagle.alert.engine.coordinator.StreamDefinition;
@@ -25,6 +24,7 @@ import org.apache.eagle.alert.engine.evaluator.PolicyStreamHandler;
 import org.apache.eagle.alert.engine.model.AlertStreamEvent;
 import org.apache.eagle.alert.engine.model.StreamEvent;
 import org.apache.eagle.alert.utils.TimePeriodUtils;
+import org.apache.commons.collections.CollectionUtils;
 import org.joda.time.Period;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -39,56 +39,54 @@ import java.util.*;
  * 2. timestamp field: timestamp column
  * 3. wiri safe time window: how long window is good for full set of wiri
  * 4. wisb: full set
- *
  * No data policy definition should include
  * fixed fields and dynamic fields
  * fixed fields are leading fields : windowPeriod, type, numOfFields, f1_name, f2_name
  * dynamic fields depend on wisb type.
- *
  * policy would be like:
  * {
- "name": "noDataAlertPolicy",
- "description": "noDataAlertPolicy",
- "inputStreams": [
- "noDataAlertStream"
- ],
- "outputStreams": [
- "noDataAlertStream_out"
- ],
- "definition": {
- "type": "nodataalert",
- "value": "PT1M,plain,1,host,host1,host2"   // or "value": "PT1M,dynamic,1,host"
- },
- "partitionSpec": [
- {
- "streamId": "noDataAlertStream",
- "type": "GROUPBY"
- }
- ],
- "parallelismHint": 2
- }
-     "name": "noDataAlertPolicy",
-     "description": "noDataAlertPolicy",
-     "inputStreams": [
-        "noDataAlertStream"
-     ],
-     "outputStreams": [
-        "noDataAlertStream_out"
-     ],
-     "definition": {
-        "type": "nodataalert",
-        "value": "PT1M,plain,1,host,host1,host2"   // or "value": "PT1M,dynamic,1,host"
-     },
-     "partitionSpec": [
-     {
-        "streamId": "noDataAlertStream",
-        "type": "GROUPBY"
-     }
-     ],
-     "parallelismHint": 2
-   }
+ * "name": "noDataAlertPolicy",
+ * "description": "noDataAlertPolicy",
+ * "inputStreams": [
+ * "noDataAlertStream"
+ * ],
+ * "outputStreams": [
+ * "noDataAlertStream_out"
+ * ],
+ * "definition": {
+ * "type": "nodataalert",
+ * "value": "PT1M,plain,1,host,host1,host2"   // or "value": "PT1M,dynamic,1,host"
+ * },
+ * "partitionSpec": [
+ * {
+ * "streamId": "noDataAlertStream",
+ * "type": "GROUPBY"
+ * }
+ * ],
+ * "parallelismHint": 2
+ * }
+ * "name": "noDataAlertPolicy",
+ * "description": "noDataAlertPolicy",
+ * "inputStreams": [
+ * "noDataAlertStream"
+ * ],
+ * "outputStreams": [
+ * "noDataAlertStream_out"
+ * ],
+ * "definition": {
+ * "type": "nodataalert",
+ * "value": "PT1M,plain,1,host,host1,host2"   // or "value": "PT1M,dynamic,1,host"
+ * },
+ * "partitionSpec": [
+ * {
+ * "streamId": "noDataAlertStream",
+ * "type": "GROUPBY"
+ * }
+ * ],
+ * "parallelismHint": 2
+ * }
  */
-public class NoDataPolicyHandler implements PolicyStreamHandler{
+public class NoDataPolicyHandler implements PolicyStreamHandler {
     private static final Logger LOG = LoggerFactory.getLogger(NoDataPolicyHandler.class);
     private Map<String, StreamDefinition> sds;
 
@@ -103,9 +101,10 @@ public class NoDataPolicyHandler implements PolicyStreamHandler{
     private volatile NoDataWisbType wisbType;
     private volatile DistinctValuesInTimeWindow distinctWindow;
 
-    public NoDataPolicyHandler(Map<String, StreamDefinition> sds){
+    public NoDataPolicyHandler(Map<String, StreamDefinition> sds) {
         this.sds = sds;
     }
+
     @Override
     public void prepare(Collector<AlertStreamEvent> collector, PolicyHandlerContext context) throws Exception {
         this.collector = collector;
@@ -113,11 +112,13 @@ public class NoDataPolicyHandler implements PolicyStreamHandler{
         this.policyDef = context.getPolicyDefinition();
         List<String> inputStreams = policyDef.getInputStreams();
         // validate inputStreams has to contain only one stream
-        if(inputStreams.size() != 1)
+        if (inputStreams.size() != 1) {
             throw new IllegalArgumentException("policy inputStream size has to be 1 for no data alert");
+        }
         // validate outputStream has to contain only one stream
-        if(policyDef.getOutputStreams().size() != 1)
+        if (policyDef.getOutputStreams().size() != 1) {
             throw new IllegalArgumentException("policy outputStream size has to be 1 for no data alert");
+        }
 
         String is = inputStreams.get(0);
         StreamDefinition sd = sds.get(is);
@@ -129,23 +130,23 @@ public class NoDataPolicyHandler implements PolicyStreamHandler{
         distinctWindow = new DistinctValuesInTimeWindow(windowPeriod);
         this.wisbType = NoDataWisbType.valueOf(segments[1]);
         // for provided wisb values, need to parse, for dynamic wisb values, it is computed through a window
-        if(wisbType == NoDataWisbType.provided) {
+        if (wisbType == NoDataWisbType.provided) {
             wisbValues = new NoDataWisbProvidedParser().parse(segments);
         }
         // populate wisb field names
         int numOfFields = Integer.parseInt(segments[2]);
-        for(int i = 3; i < 3+numOfFields; i++){
+        for (int i = 3; i < 3 + numOfFields; i++) {
             String fn = segments[i];
             wisbFieldIndices.add(sd.getColumnIndex(fn));
         }
     }
 
-    @SuppressWarnings({ "rawtypes", "unchecked" })
+    @SuppressWarnings( {"rawtypes", "unchecked"})
     @Override
     public void send(StreamEvent event) throws Exception {
         Object[] data = event.getData();
         List<Object> columnValues = new ArrayList<>();
-        for(int i=0; i<wisbFieldIndices.size(); i++){
+        for (int i = 0; i < wisbFieldIndices.size(); i++) {
             Object o = data[wisbFieldIndices.get(i)];
             // convert value to string
             columnValues.add(o.toString());
@@ -155,18 +156,18 @@ public class NoDataPolicyHandler implements PolicyStreamHandler{
 
         LOG.debug("window slided: {}, with wiri: {}", distinctWindow.windowSlided(), distinctWindow.distinctValues());
 
-        if(distinctWindow.windowSlided()) {
+        if (distinctWindow.windowSlided()) {
             compareAndEmit(wisbValues, wiriValues, event);
         }
 
-        if(wisbType == NoDataWisbType.dynamic) {
+        if (wisbType == NoDataWisbType.dynamic) {
             // deep copy
             wisbValues = new HashSet<>(wiriValues);
         }
     }
 
     @SuppressWarnings("rawtypes")
-    private void compareAndEmit(Set wisb, Set wiri, StreamEvent event){
+    private void compareAndEmit(Set wisb, Set wiri, StreamEvent event) {
         // compare with wisbValues if wisbValues are already there for dynamic type
         Collection noDataValues = CollectionUtils.subtract(wisb, wiri);
         LOG.debug("nodatavalues:" + noDataValues + ", wisb: " + wisb + ", wiri: " + wiri);
@@ -177,9 +178,9 @@ public class NoDataPolicyHandler implements PolicyStreamHandler{
         }
     }
 
-    private AlertStreamEvent createAlertEvent(long timestamp, Object[] triggerEvent){
+    private AlertStreamEvent createAlertEvent(long timestamp, Object[] triggerEvent) {
         String is = policyDef.getInputStreams().get(0);
-        StreamDefinition sd = sds.get(is);
+        final StreamDefinition sd = sds.get(is);
 
         AlertStreamEvent event = new AlertStreamEvent();
         event.setTimestamp(timestamp);

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/a0fc8708/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/nodata/NoDataPolicyTimeBatchHandler.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/nodata/NoDataPolicyTimeBatchHandler.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/nodata/NoDataPolicyTimeBatchHandler.java
index 53a7af0..73ee9b2 100644
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/nodata/NoDataPolicyTimeBatchHandler.java
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/nodata/NoDataPolicyTimeBatchHandler.java
@@ -16,7 +16,6 @@
  */
 package org.apache.eagle.alert.engine.evaluator.nodata;
 
-import org.apache.commons.collections.CollectionUtils;
 import org.apache.eagle.alert.engine.Collector;
 import org.apache.eagle.alert.engine.coordinator.PolicyDefinition;
 import org.apache.eagle.alert.engine.coordinator.StreamDefinition;
@@ -25,6 +24,7 @@ import org.apache.eagle.alert.engine.evaluator.PolicyStreamHandler;
 import org.apache.eagle.alert.engine.model.AlertStreamEvent;
 import org.apache.eagle.alert.engine.model.StreamEvent;
 import org.apache.eagle.alert.utils.TimePeriodUtils;
+import org.apache.commons.collections.CollectionUtils;
 import org.apache.storm.guava.base.Joiner;
 import org.joda.time.Period;
 import org.slf4j.Logger;
@@ -33,132 +33,134 @@ import org.slf4j.LoggerFactory;
 import java.util.*;
 
 public class NoDataPolicyTimeBatchHandler implements PolicyStreamHandler {
-	
-	private static final Logger LOG = LoggerFactory.getLogger(NoDataPolicyTimeBatchHandler.class);
-	private Map<String, StreamDefinition> sds;
-	
-	private volatile List<Integer> wisbFieldIndices = new ArrayList<>();
-	// reuse PolicyDefinition.defintion.value field to store full set of values
-	// separated by comma
-	private volatile PolicyDefinition policyDef;
-	private volatile Collector<AlertStreamEvent> collector;
-	private volatile PolicyHandlerContext context;
-	private volatile NoDataWisbType wisbType;
-	private volatile DistinctValuesInTimeBatchWindow distinctWindow;
-
-	public NoDataPolicyTimeBatchHandler(Map<String, StreamDefinition> sds){
+
+    private static final Logger LOG = LoggerFactory.getLogger(NoDataPolicyTimeBatchHandler.class);
+    private Map<String, StreamDefinition> sds;
+
+    private volatile List<Integer> wisbFieldIndices = new ArrayList<>();
+    // reuse PolicyDefinition.defintion.value field to store full set of values
+    // separated by comma
+    private volatile PolicyDefinition policyDef;
+    private volatile Collector<AlertStreamEvent> collector;
+    private volatile PolicyHandlerContext context;
+    private volatile NoDataWisbType wisbType;
+    private volatile DistinctValuesInTimeBatchWindow distinctWindow;
+
+    public NoDataPolicyTimeBatchHandler(Map<String, StreamDefinition> sds) {
         this.sds = sds;
     }
 
-	@Override
-	public void prepare(Collector<AlertStreamEvent> collector, PolicyHandlerContext context) throws Exception {
-		this.collector = collector;
-		this.context = context;
-		this.policyDef = context.getPolicyDefinition();
-		List<String> inputStreams = policyDef.getInputStreams();
-		// validate inputStreams has to contain only one stream
-		if (inputStreams.size() != 1)
-			throw new IllegalArgumentException("policy inputStream size has to be 1 for no data alert");
-		// validate outputStream has to contain only one stream
-		if (policyDef.getOutputStreams().size() != 1)
-			throw new IllegalArgumentException("policy outputStream size has to be 1 for no data alert");
-
-		String is = inputStreams.get(0);
-		StreamDefinition sd = sds.get(is);
-
-		String policyValue = policyDef.getDefinition().getValue();
-		// assume that no data alert policy value consists of "windowPeriod,
-		// type, numOfFields, f1_name, f2_name, f1_value, f2_value, f1_value,
-		// f2_value}
-		String[] segments = policyValue.split(",");
-		this.wisbType = NoDataWisbType.valueOf(segments[1]);
-		// for provided wisb values, need to parse, for dynamic wisb values, it
-		// is computed through a window
-		@SuppressWarnings("rawtypes")
-		Set wisbValues = null;
-		if (wisbType == NoDataWisbType.provided) {
-			wisbValues = new NoDataWisbProvidedParser().parse(segments);
-		}
-		long windowPeriod = TimePeriodUtils.getMillisecondsOfPeriod(Period.parse(segments[0]));
-		distinctWindow = new DistinctValuesInTimeBatchWindow(this, windowPeriod, wisbValues);
-		// populate wisb field names
-		int numOfFields = Integer.parseInt(segments[2]);
-		for (int i = 3; i < 3 + numOfFields; i++) {
-			String fn = segments[i];
-			wisbFieldIndices.add(sd.getColumnIndex(fn));
-		}
-	}
-
-	@Override
-	public void send(StreamEvent event) throws Exception {
-		Object[] data = event.getData();
-		
-		List<Object> columnValues = new ArrayList<>();
-		for (int i = 0; i < wisbFieldIndices.size(); i++) {
-			Object o = data[wisbFieldIndices.get(i)];
-			// convert value to string
-			columnValues.add(o.toString());
-		}
-		// use local timestamp rather than event timestamp
-		distinctWindow.send(event, columnValues, System.currentTimeMillis());
-		LOG.debug("event sent to window with wiri: {}", distinctWindow.distinctValues());
-	}
-	
-	@SuppressWarnings("rawtypes")
-	public void compareAndEmit(Set wisb, Set wiri, StreamEvent event) {
-		// compare with wisbValues if wisbValues are already there for dynamic
-		// type
-		Collection noDataValues = CollectionUtils.subtract(wisb, wiri);
-		LOG.debug("nodatavalues:" + noDataValues + ", wisb: " + wisb + ", wiri: " + wiri);
-		if (noDataValues != null && noDataValues.size() > 0) {
-			LOG.info("No data alert is triggered with no data values {} and wisb {}", noDataValues, wisb);
-			
-			String is = policyDef.getOutputStreams().get(0);
-			StreamDefinition sd = sds.get(is);
-			int timestampIndex = sd.getColumnIndex("timestamp");
-			int hostIndex = sd.getColumnIndex("host");
-			int originalStreamNameIndex = sd.getColumnIndex("originalStreamName");
-			
-			for (Object one : noDataValues) {
-				Object[] triggerEvent = new Object[sd.getColumns().size()];
-				for (int i = 0; i < sd.getColumns().size(); i ++) {
-					if (i == timestampIndex) {
-						triggerEvent[i] = System.currentTimeMillis();
-					} else if (i == hostIndex) {
-						triggerEvent[hostIndex] = ((List) one).get(0);
-					} else if (i == originalStreamNameIndex) {
-						triggerEvent[originalStreamNameIndex] = event.getStreamId();
-					} else if (sd.getColumns().size() < i) {
-						LOG.error("strema event data have different lenght compare to column definition!");
-					} else {
-						triggerEvent[i] = sd.getColumns().get(i).getDefaultValue();
-					}
-				}
-				AlertStreamEvent alertEvent = createAlertEvent(sd, event.getTimestamp(), triggerEvent);
-				LOG.info(String.format("Nodata alert %s generated and will be emitted", Joiner.on(",").join(triggerEvent)));
-				collector.emit(alertEvent);
-			}
-			
-		}
-	}
-
-	private AlertStreamEvent createAlertEvent(StreamDefinition sd, long timestamp, Object[] triggerEvent) {
-		AlertStreamEvent event = new AlertStreamEvent();
-		event.setTimestamp(timestamp);
-		event.setData(triggerEvent);
-		event.setStreamId(policyDef.getOutputStreams().get(0));
-		event.setPolicyId(context.getPolicyDefinition().getName());
-		if (this.context.getPolicyEvaluator() != null) {
-			event.setCreatedBy(context.getPolicyEvaluator().getName());
-		}
-		event.setCreatedTime(System.currentTimeMillis());
-		event.setSchema(sd);
-		return event;
-	}
-
-	@Override
-	public void close() throws Exception {
-
-	}
+    @Override
+    public void prepare(Collector<AlertStreamEvent> collector, PolicyHandlerContext context) throws Exception {
+        this.collector = collector;
+        this.context = context;
+        this.policyDef = context.getPolicyDefinition();
+        List<String> inputStreams = policyDef.getInputStreams();
+        // validate inputStreams has to contain only one stream
+        if (inputStreams.size() != 1) {
+            throw new IllegalArgumentException("policy inputStream size has to be 1 for no data alert");
+        }
+        // validate outputStream has to contain only one stream
+        if (policyDef.getOutputStreams().size() != 1) {
+            throw new IllegalArgumentException("policy outputStream size has to be 1 for no data alert");
+        }
+
+        String is = inputStreams.get(0);
+        StreamDefinition sd = sds.get(is);
+
+        String policyValue = policyDef.getDefinition().getValue();
+        // assume that no data alert policy value consists of "windowPeriod,
+        // type, numOfFields, f1_name, f2_name, f1_value, f2_value, f1_value,
+        // f2_value}
+        String[] segments = policyValue.split(",");
+        this.wisbType = NoDataWisbType.valueOf(segments[1]);
+        // for provided wisb values, need to parse, for dynamic wisb values, it
+        // is computed through a window
+        @SuppressWarnings("rawtypes")
+        Set wisbValues = null;
+        if (wisbType == NoDataWisbType.provided) {
+            wisbValues = new NoDataWisbProvidedParser().parse(segments);
+        }
+        long windowPeriod = TimePeriodUtils.getMillisecondsOfPeriod(Period.parse(segments[0]));
+        distinctWindow = new DistinctValuesInTimeBatchWindow(this, windowPeriod, wisbValues);
+        // populate wisb field names
+        int numOfFields = Integer.parseInt(segments[2]);
+        for (int i = 3; i < 3 + numOfFields; i++) {
+            String fn = segments[i];
+            wisbFieldIndices.add(sd.getColumnIndex(fn));
+        }
+    }
+
+    @Override
+    public void send(StreamEvent event) throws Exception {
+        Object[] data = event.getData();
+
+        List<Object> columnValues = new ArrayList<>();
+        for (int i = 0; i < wisbFieldIndices.size(); i++) {
+            Object o = data[wisbFieldIndices.get(i)];
+            // convert value to string
+            columnValues.add(o.toString());
+        }
+        // use local timestamp rather than event timestamp
+        distinctWindow.send(event, columnValues, System.currentTimeMillis());
+        LOG.debug("event sent to window with wiri: {}", distinctWindow.distinctValues());
+    }
+
+    @SuppressWarnings("rawtypes")
+    public void compareAndEmit(Set wisb, Set wiri, StreamEvent event) {
+        // compare with wisbValues if wisbValues are already there for dynamic
+        // type
+        Collection noDataValues = CollectionUtils.subtract(wisb, wiri);
+        LOG.debug("nodatavalues:" + noDataValues + ", wisb: " + wisb + ", wiri: " + wiri);
+        if (noDataValues != null && noDataValues.size() > 0) {
+            LOG.info("No data alert is triggered with no data values {} and wisb {}", noDataValues, wisb);
+
+            String is = policyDef.getOutputStreams().get(0);
+            StreamDefinition sd = sds.get(is);
+            int timestampIndex = sd.getColumnIndex("timestamp");
+            int hostIndex = sd.getColumnIndex("host");
+            int originalStreamNameIndex = sd.getColumnIndex("originalStreamName");
+
+            for (Object one : noDataValues) {
+                Object[] triggerEvent = new Object[sd.getColumns().size()];
+                for (int i = 0; i < sd.getColumns().size(); i++) {
+                    if (i == timestampIndex) {
+                        triggerEvent[i] = System.currentTimeMillis();
+                    } else if (i == hostIndex) {
+                        triggerEvent[hostIndex] = ((List) one).get(0);
+                    } else if (i == originalStreamNameIndex) {
+                        triggerEvent[originalStreamNameIndex] = event.getStreamId();
+                    } else if (sd.getColumns().size() < i) {
+                        LOG.error("strema event data have different lenght compare to column definition!");
+                    } else {
+                        triggerEvent[i] = sd.getColumns().get(i).getDefaultValue();
+                    }
+                }
+                AlertStreamEvent alertEvent = createAlertEvent(sd, event.getTimestamp(), triggerEvent);
+                LOG.info(String.format("Nodata alert %s generated and will be emitted", Joiner.on(",").join(triggerEvent)));
+                collector.emit(alertEvent);
+            }
+
+        }
+    }
+
+    private AlertStreamEvent createAlertEvent(StreamDefinition sd, long timestamp, Object[] triggerEvent) {
+        AlertStreamEvent event = new AlertStreamEvent();
+        event.setTimestamp(timestamp);
+        event.setData(triggerEvent);
+        event.setStreamId(policyDef.getOutputStreams().get(0));
+        event.setPolicyId(context.getPolicyDefinition().getName());
+        if (this.context.getPolicyEvaluator() != null) {
+            event.setCreatedBy(context.getPolicyEvaluator().getName());
+        }
+        event.setCreatedTime(System.currentTimeMillis());
+        event.setSchema(sd);
+        return event;
+    }
+
+    @Override
+    public void close() throws Exception {
+
+    }
 
 }

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/a0fc8708/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/nodata/NoDataWisbParser.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/nodata/NoDataWisbParser.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/nodata/NoDataWisbParser.java
index fe06067..fa27108 100644
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/nodata/NoDataWisbParser.java
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/nodata/NoDataWisbParser.java
@@ -25,7 +25,8 @@ import java.util.Set;
 public interface NoDataWisbParser {
     /**
      * parse policy definition and return WISB values for one or multiple fields
-     * for example host and data center are 2 fields for no data alert, then WISB is a list of two values
+     * for example host and data center are 2 fields for no data alert, then WISB is a list of two values.
+     *
      * @param args some information parsed from policy defintion
      * @return list of list of field values
      */

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/a0fc8708/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/nodata/NoDataWisbProvidedParser.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/nodata/NoDataWisbProvidedParser.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/nodata/NoDataWisbProvidedParser.java
index e13826a..4f54358 100644
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/nodata/NoDataWisbProvidedParser.java
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/nodata/NoDataWisbProvidedParser.java
@@ -24,19 +24,19 @@ import java.util.Set;
 /**
  * Since 6/29/16.
  */
-public class NoDataWisbProvidedParser implements NoDataWisbParser{
-    @Override
+public class NoDataWisbProvidedParser implements NoDataWisbParser {
     /**
-     * policy value consists of "windowPeriod, type, numOfFields, f1_name, f2_name, f1_value, f2_value, f1_value, f2_value"
+     * policy value consists of "windowPeriod, type, numOfFields, f1_name, f2_name, f1_value, f2_value, f1_value, f2_value".
      */
+    @Override
     public Set<List<String>> parse(String[] args) {
         int numOfFields = Integer.parseInt(args[2]);
         Set<List<String>> wisbValues = new HashSet<>();
         int i = 3 + numOfFields;
-        while(i<args.length){
+        while (i < args.length) {
             List<String> fields = new ArrayList<>();
-            for(int j=0; j<numOfFields; j++){
-                fields.add(args[i+j]);
+            for (int j = 0; j < numOfFields; j++) {
+                fields.add(args[i + j]);
             }
             wisbValues.add(fields);
             i += numOfFields;

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/a0fc8708/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertDeduplicator.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertDeduplicator.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertDeduplicator.java
index ed04d5a..2f71c7f 100644
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertDeduplicator.java
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertDeduplicator.java
@@ -25,8 +25,8 @@ import org.apache.eagle.alert.engine.model.AlertStreamEvent;
  */
 public interface AlertDeduplicator {
 
-	AlertStreamEvent dedup(AlertStreamEvent event);
+    AlertStreamEvent dedup(AlertStreamEvent event);
 
-	void setDedupIntervalMin(String intervalMin);
+    void setDedupIntervalMin(String intervalMin);
 
 }

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/a0fc8708/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertPublishListener.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertPublishListener.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertPublishListener.java
index 8f1c248..fdb01a7 100644
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertPublishListener.java
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertPublishListener.java
@@ -1,10 +1,4 @@
-package org.apache.eagle.alert.engine.publisher;
-
-import java.util.List;
-
-import org.apache.eagle.alert.engine.coordinator.Publishment;
-
-/**
+/*
  * 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.
@@ -20,6 +14,12 @@ import org.apache.eagle.alert.engine.coordinator.Publishment;
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+package org.apache.eagle.alert.engine.publisher;
+
+import org.apache.eagle.alert.engine.coordinator.Publishment;
+
+import java.util.List;
+
 public interface AlertPublishListener {
     void onPublishChange(List<Publishment> added,
                          List<Publishment> removed,

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/a0fc8708/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertPublishPlugin.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertPublishPlugin.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertPublishPlugin.java
index d24bdb0..4c3a2ad 100644
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertPublishPlugin.java
+++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/publisher/AlertPublishPlugin.java
@@ -17,22 +17,22 @@
  */
 package org.apache.eagle.alert.engine.publisher;
 
-import java.io.Closeable;
-import java.util.Map;
-
 import org.apache.eagle.alert.engine.coordinator.Publishment;
 import org.apache.eagle.alert.engine.model.AlertStreamEvent;
 import org.apache.eagle.alert.engine.publisher.impl.PublishStatus;
-
 import com.typesafe.config.Config;
 
+import java.io.Closeable;
+import java.util.Map;
+
 /**
  * Created on 2/10/16.
  * Notification Plug-in interface which provide abstraction layer to notify to different system
  */
 public interface AlertPublishPlugin extends Closeable {
     /**
-     * 
+     * Init alert publish plugin.
+     *
      * @param config
      * @param publishment
      * @param configProperties - storm config that would be useful for some implementation



Mime
View raw message