eagle-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From h..@apache.org
Subject [26/84] [partial] eagle git commit: Clean repo for eagle site
Date Mon, 03 Apr 2017 11:54:34 GMT
http://git-wip-us.apache.org/repos/asf/eagle/blob/6fd95d5c/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
deleted file mode 100755
index 628b2e4..0000000
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/SiddhiPolicyHandler.java
+++ /dev/null
@@ -1,123 +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.alert.engine.evaluator.impl;
-
-import org.apache.eagle.alert.engine.Collector;
-import org.apache.eagle.alert.engine.coordinator.PolicyDefinition;
-import org.apache.eagle.alert.engine.coordinator.StreamDefinition;
-import org.apache.eagle.alert.engine.coordinator.StreamNotDefinedException;
-import org.apache.eagle.alert.engine.evaluator.PolicyHandlerContext;
-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.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.wso2.siddhi.core.ExecutionPlanRuntime;
-import org.wso2.siddhi.core.SiddhiManager;
-import org.wso2.siddhi.core.stream.input.InputHandler;
-
-import java.util.List;
-import java.util.Map;
-
-public class SiddhiPolicyHandler implements PolicyStreamHandler {
-    private static final Logger LOG = LoggerFactory.getLogger(SiddhiPolicyHandler.class);
-    private ExecutionPlanRuntime executionRuntime;
-    private SiddhiManager siddhiManager;
-    private Map<String, StreamDefinition> sds;
-    private PolicyDefinition policy;
-    private PolicyHandlerContext context;
-
-    private int currentIndex = 0; // the index of current definition statement inside the policy definition
-
-    public SiddhiPolicyHandler(Map<String, StreamDefinition> sds, int index) {
-        this.sds = sds;
-        this.currentIndex = index;
-    }
-
-    protected String generateExecutionPlan(PolicyDefinition policyDefinition, Map<String, StreamDefinition> sds) throws StreamNotDefinedException {
-        return SiddhiDefinitionAdapter.buildSiddhiExecutionPlan(policyDefinition,sds);
-    }
-
-    @Override
-    public void prepare(final Collector<AlertStreamEvent> collector, PolicyHandlerContext context) throws Exception {
-        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);
-            throw parserException;
-        }
-
-        // add output stream callback
-        List<String> outputStreams = getOutputStreams(policy);
-        for (final String outputStream : outputStreams) {
-            if (executionRuntime.getStreamDefinitionMap().containsKey(outputStream)) {
-                StreamDefinition streamDefinition = SiddhiDefinitionAdapter.convertFromSiddiDefinition(executionRuntime.getStreamDefinitionMap().get(outputStream));
-                this.executionRuntime.addCallback(outputStream,
-                    new AlertStreamCallback(outputStream, streamDefinition,
-                        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());
-    }
-
-    protected List<String> getOutputStreams(PolicyDefinition policy) {
-        return policy.getOutputStreams().isEmpty() ? policy.getDefinition().getOutputStreams() : policy.getOutputStreams();
-    }
-
-    public void send(StreamEvent event) throws Exception {
-        context.getPolicyCounter().incr(String.format("%s.%s", this.context.getPolicyDefinition().getName(), "receive_count"));
-        String streamId = event.getStreamId();
-        InputHandler inputHandler = executionRuntime.getInputHandler(streamId);
-        if (inputHandler != null) {
-            context.getPolicyCounter().incr(String.format("%s.%s", this.context.getPolicyDefinition().getName(), "eval_count"));
-            inputHandler.send(event.getTimestamp(), event.getData());
-
-            if (LOG.isDebugEnabled()) {
-                LOG.debug("sent event to siddhi stream {} ", streamId);
-            }
-        } else {
-            context.getPolicyCounter().incr(String.format("%s.%s", this.context.getPolicyDefinition().getName(), "drop_count"));
-            LOG.warn("No input handler found for stream {}", streamId);
-        }
-    }
-
-    public void close() throws Exception {
-        LOG.info("Closing handler for policy {}", this.policy.getName());
-        this.executionRuntime.shutdown();
-        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());
-    }
-
-    @Override
-    public String toString() {
-        StringBuilder sb = new StringBuilder("SiddhiPolicyHandler for policy: ");
-        sb.append(this.policy == null ? "" : this.policy.getName());
-        return sb.toString();
-    }
-
-}

http://git-wip-us.apache.org/repos/asf/eagle/blob/6fd95d5c/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
deleted file mode 100644
index 141c819..0000000
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/impl/SiddhiPolicyStateHandler.java
+++ /dev/null
@@ -1,62 +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.alert.engine.evaluator.impl;
-
-import org.apache.eagle.alert.engine.coordinator.PolicyDefinition;
-import org.apache.eagle.alert.engine.coordinator.StreamDefinition;
-import org.apache.eagle.alert.engine.coordinator.StreamNotDefinedException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.List;
-import java.util.Map;
-
-/**
- * Created on 7/27/16.
- */
-public class SiddhiPolicyStateHandler extends SiddhiPolicyHandler {
-
-    private static final Logger LOG = LoggerFactory.getLogger(SiddhiPolicyStateHandler.class);
-
-    public SiddhiPolicyStateHandler(Map<String, StreamDefinition> sds, int index) {
-        super(sds, index);
-    }
-
-    @Override
-    protected String generateExecutionPlan(PolicyDefinition policyDefinition, Map<String, StreamDefinition> sds) throws StreamNotDefinedException {
-        StringBuilder builder = new StringBuilder();
-        PolicyDefinition.Definition stateDefiniton = policyDefinition.getStateDefinition();
-        List<String> inputStreams = stateDefiniton.getInputStreams();
-        for (String inputStream : inputStreams) { // 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);
-        }
-        return builder.toString();
-    }
-
-    @Override
-    protected List<String> getOutputStreams(PolicyDefinition policy) {
-        return policy.getStateDefinition().getOutputStreams();
-    }
-
-    // more validation on prepare
-
-}

http://git-wip-us.apache.org/repos/asf/eagle/blob/6fd95d5c/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
deleted file mode 100644
index ef806fb..0000000
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/nodata/DistinctValuesInTimeBatchWindow.java
+++ /dev/null
@@ -1,128 +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
- *
- *    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.
- */
-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;
-import java.util.Set;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
-
-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);
-    }
-
-}

http://git-wip-us.apache.org/repos/asf/eagle/blob/6fd95d5c/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
deleted file mode 100644
index 4aae040..0000000
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/nodata/DistinctValuesInTimeWindow.java
+++ /dev/null
@@ -1,140 +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.alert.engine.evaluator.nodata;
-
-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).
- */
-public class DistinctValuesInTimeWindow {
-    public static class ValueAndTime {
-        Object value;
-        long timestamp;
-
-        public ValueAndTime(Object value, long timestamp) {
-            this.value = value;
-            this.timestamp = timestamp;
-        }
-
-        public String toString() {
-            return "[" + value + "," + timestamp + "]";
-        }
-
-        public int hashCode() {
-            return new HashCodeBuilder().append(value).append(timestamp).toHashCode();
-        }
-
-        public boolean equals(Object that) {
-            if (!(that instanceof ValueAndTime)) {
-                return false;
-            }
-            ValueAndTime another = (ValueAndTime) that;
-            return another.timestamp == this.timestamp && another.value.equals(this.value);
-        }
-    }
-
-    public static class ValueAndTimeComparator implements Comparator<ValueAndTime> {
-        @Override
-        public int compare(ValueAndTime o1, ValueAndTime o2) {
-            if (o1.timestamp != o2.timestamp) {
-                return (o1.timestamp > o2.timestamp) ? 1 : -1;
-            }
-            if (o1.value.equals(o2.value)) {
-                return 0;
-            } else {
-                // this is not strictly correct, but I don't want to write too many comparators here :-)
-                if (o1.hashCode() > o2.hashCode()) {
-                    return 1;
-                } else {
-                    return -1;
-                }
-            }
-        }
-    }
-
-    /**
-     * 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.
-     */
-    private SortedMap<ValueAndTime, ValueAndTime> timeSortedMap = new TreeMap<>(new ValueAndTimeComparator());
-    private long maxTimestamp = 0L;
-    private long window;
-    private boolean windowSlided;
-
-    /**
-     * @param window - milliseconds.
-     */
-    public DistinctValuesInTimeWindow(long window) {
-        this.window = window;
-    }
-
-    public void send(Object value, long timestamp) {
-        ValueAndTime vt = new ValueAndTime(value, timestamp);
-
-        // todo think of time out of order
-        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;
-            }
-            timeSortedMap.remove(new ValueAndTime(value, oldTime));
-        }
-        // insert entry with new timestamp in timeSortedMap
-        timeSortedMap.put(vt, vt);
-        // update new timestamp in valueMaxTimeMap
-        valueMaxTimeMap.put(value, timestamp);
-
-        // evict old entries
-        // store max timestamp if possible
-        maxTimestamp = Math.max(maxTimestamp, timestamp);
-
-        // check if some values should be evicted because of time window
-        Iterator<Map.Entry<ValueAndTime, ValueAndTime>> it = timeSortedMap.entrySet().iterator();
-        while (it.hasNext()) {
-            Map.Entry<ValueAndTime, ValueAndTime> entry = it.next();
-            if (entry.getKey().timestamp < maxTimestamp - window) {
-                // should remove the entry in valueMaxTimeMap and timeSortedMap
-                valueMaxTimeMap.remove(entry.getKey().value);
-                windowSlided = true;
-
-                it.remove();
-            } else {
-                break;
-            }
-        }
-    }
-
-    public Map<Object, Long> distinctValues() {
-        return valueMaxTimeMap;
-    }
-
-    public boolean windowSlided() {
-        return windowSlided;
-    }
-}

http://git-wip-us.apache.org/repos/asf/eagle/blob/6fd95d5c/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
deleted file mode 100644
index ec6e6e9..0000000
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/nodata/NoDataPolicyHandler.java
+++ /dev/null
@@ -1,202 +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.alert.engine.evaluator.nodata;
-
-import org.apache.eagle.alert.engine.Collector;
-import org.apache.eagle.alert.engine.coordinator.PolicyDefinition;
-import org.apache.eagle.alert.engine.coordinator.StreamDefinition;
-import org.apache.eagle.alert.engine.evaluator.PolicyHandlerContext;
-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;
-
-import java.util.*;
-
-/**
- * Since 6/28/16.
- * No Data Policy engine
- * based on the following information
- * 1. stream definition: group by columns
- * 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
- * }
- */
-public class NoDataPolicyHandler implements PolicyStreamHandler {
-    private static final Logger LOG = LoggerFactory.getLogger(NoDataPolicyHandler.class);
-    private Map<String, StreamDefinition> sds;
-
-    // wisb(what is should be) set for expected full set value of multiple columns
-    @SuppressWarnings("rawtypes")
-    private volatile Set wisbValues = null;
-    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 DistinctValuesInTimeWindow distinctWindow;
-
-    public NoDataPolicyHandler(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(",");
-        long windowPeriod = TimePeriodUtils.getMillisecondsOfPeriod(Period.parse(segments[0]));
-        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) {
-            wisbValues = new NoDataWisbProvidedParser().parse(segments);
-        }
-        // 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));
-        }
-    }
-
-    @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++) {
-            Object o = data[wisbFieldIndices.get(i)];
-            // convert value to string
-            columnValues.add(o.toString());
-        }
-        distinctWindow.send(columnValues, event.getTimestamp());
-        Set wiriValues = distinctWindow.distinctValues().keySet();
-
-        LOG.debug("window slided: {}, with wiri: {}", distinctWindow.windowSlided(), distinctWindow.distinctValues());
-
-        if (distinctWindow.windowSlided()) {
-            compareAndEmit(wisbValues, wiriValues, event);
-        }
-
-        if (wisbType == NoDataWisbType.dynamic) {
-            // deep copy
-            wisbValues = new HashSet<>(wiriValues);
-        }
-    }
-
-    @SuppressWarnings("rawtypes")
-    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);
-        if (noDataValues != null && noDataValues.size() > 0) {
-            LOG.info("No data alert is triggered with no data values {} and wisb {}", noDataValues, wisbValues);
-            AlertStreamEvent alertEvent = createAlertEvent(event.getTimestamp(), event.getData());
-            collector.emit(alertEvent);
-        }
-    }
-
-    private AlertStreamEvent createAlertEvent(long timestamp, Object[] triggerEvent) {
-        String is = policyDef.getInputStreams().get(0);
-        final StreamDefinition sd = sds.get(is);
-
-        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 {
-
-    }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/eagle/blob/6fd95d5c/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
deleted file mode 100644
index b1e32bd..0000000
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/nodata/NoDataPolicyTimeBatchHandler.java
+++ /dev/null
@@ -1,173 +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
- *
- *    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.
- */
-package org.apache.eagle.alert.engine.evaluator.nodata;
-
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-
-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;
-import org.apache.eagle.alert.engine.evaluator.PolicyHandlerContext;
-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.joda.time.Period;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.google.common.base.Joiner;
-
-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) {
-        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 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
-        Set<String> wisbValues = new HashSet<String>();
-        if (wisbType == NoDataWisbType.provided) {
-            for (int i = 2; i < segments.length; i++) {
-                wisbValues.add(segments[i]);
-            }
-        }
-        
-        long windowPeriod = TimePeriodUtils.getMillisecondsOfPeriod(Period.parse(segments[0]));
-        distinctWindow = new DistinctValuesInTimeBatchWindow(this, windowPeriod, wisbValues);
-        // populate wisb field names
-        String is = inputStreams.get(0);
-        StreamDefinition sd = sds.get(is);
-        String nodataColumnNameKey = "nodataColumnName";
-        if (!policyDef.getDefinition().getProperties().containsKey(nodataColumnNameKey)) {
-            throw new IllegalArgumentException("policy nodata column name has to be defined for no data alert");
-        }
-        wisbFieldIndices.add(sd.getColumnIndex((String) policyDef.getDefinition().getProperties().get(nodataColumnNameKey)));
-    }
-
-    @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/eagle/blob/6fd95d5c/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
deleted file mode 100644
index fa27108..0000000
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/nodata/NoDataWisbParser.java
+++ /dev/null
@@ -1,34 +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.alert.engine.evaluator.nodata;
-
-import java.util.List;
-import java.util.Set;
-
-/**
- * Since 6/29/16.
- */
-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.
-     *
-     * @param args some information parsed from policy defintion
-     * @return list of list of field values
-     */
-    Set<List<String>> parse(String[] args);
-}

http://git-wip-us.apache.org/repos/asf/eagle/blob/6fd95d5c/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
deleted file mode 100644
index 4f54358..0000000
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/nodata/NoDataWisbProvidedParser.java
+++ /dev/null
@@ -1,46 +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.alert.engine.evaluator.nodata;
-
-import java.util.ArrayList;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Set;
-
-/**
- * Since 6/29/16.
- */
-public class NoDataWisbProvidedParser implements NoDataWisbParser {
-    /**
-     * 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) {
-            List<String> fields = new ArrayList<>();
-            for (int j = 0; j < numOfFields; j++) {
-                fields.add(args[i + j]);
-            }
-            wisbValues.add(fields);
-            i += numOfFields;
-        }
-        return wisbValues;
-    }
-}

http://git-wip-us.apache.org/repos/asf/eagle/blob/6fd95d5c/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/nodata/NoDataWisbType.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/nodata/NoDataWisbType.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/nodata/NoDataWisbType.java
deleted file mode 100644
index 887d099..0000000
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/evaluator/nodata/NoDataWisbType.java
+++ /dev/null
@@ -1,25 +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.alert.engine.evaluator.nodata;
-
-/**
- * Since 6/29/16.
- */
-public enum NoDataWisbType {
-    provided,
-    dynamic
-}

http://git-wip-us.apache.org/repos/asf/eagle/blob/6fd95d5c/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/interpreter/PolicyExecutionPlan.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/interpreter/PolicyExecutionPlan.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/interpreter/PolicyExecutionPlan.java
deleted file mode 100644
index 7ecc36f..0000000
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/interpreter/PolicyExecutionPlan.java
+++ /dev/null
@@ -1,100 +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.alert.engine.interpreter;
-
-import com.fasterxml.jackson.databind.annotation.JsonSerialize;
-import org.apache.eagle.alert.engine.coordinator.StreamColumn;
-import org.apache.eagle.alert.engine.coordinator.StreamPartition;
-import org.wso2.siddhi.query.api.ExecutionPlan;
-
-import java.util.List;
-import java.util.Map;
-
-@JsonSerialize(include = JsonSerialize.Inclusion.NON_NULL)
-public class PolicyExecutionPlan {
-    /**
-     * Actual input streams.
-     */
-    private Map<String, List<StreamColumn>> inputStreams;
-
-    /**
-     * Actual output streams.
-     */
-    private Map<String, List<StreamColumn>> outputStreams;
-
-    /**
-     * Execution plan source.
-     */
-    private String executionPlanSource;
-
-    /**
-     * Execution plan.
-     */
-    private ExecutionPlan internalExecutionPlan;
-
-    private String executionPlanDesc;
-
-    private List<StreamPartition> streamPartitions;
-
-    public String getExecutionPlanSource() {
-        return executionPlanSource;
-    }
-
-    public void setExecutionPlanSource(String executionPlanSource) {
-        this.executionPlanSource = executionPlanSource;
-    }
-
-    public ExecutionPlan getInternalExecutionPlan() {
-        return internalExecutionPlan;
-    }
-
-    public void setInternalExecutionPlan(ExecutionPlan internalExecutionPlan) {
-        this.internalExecutionPlan = internalExecutionPlan;
-    }
-
-    public String getExecutionPlanDesc() {
-        return executionPlanDesc;
-    }
-
-    public void setExecutionPlanDesc(String executionPlanDesc) {
-        this.executionPlanDesc = executionPlanDesc;
-    }
-
-    public List<StreamPartition> getStreamPartitions() {
-        return streamPartitions;
-    }
-
-    public void setStreamPartitions(List<StreamPartition> streamPartitions) {
-        this.streamPartitions = streamPartitions;
-    }
-
-    public Map<String, List<StreamColumn>> getInputStreams() {
-        return inputStreams;
-    }
-
-    public void setInputStreams(Map<String, List<StreamColumn>> inputStreams) {
-        this.inputStreams = inputStreams;
-    }
-
-    public Map<String, List<StreamColumn>> getOutputStreams() {
-        return outputStreams;
-    }
-
-    public void setOutputStreams(Map<String, List<StreamColumn>> outputStreams) {
-        this.outputStreams = outputStreams;
-    }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/eagle/blob/6fd95d5c/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/interpreter/PolicyExecutionPlanner.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/interpreter/PolicyExecutionPlanner.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/interpreter/PolicyExecutionPlanner.java
deleted file mode 100644
index b8e5e42..0000000
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/interpreter/PolicyExecutionPlanner.java
+++ /dev/null
@@ -1,31 +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.alert.engine.interpreter;
-
-/**
- * Keep PolicyExecutionPlanner as simple and fast as possible (avoid any backend data exchanging).
- */
-interface PolicyExecutionPlanner {
-    /**
-     * @return PolicyExecutionPlan.
-     */
-    PolicyExecutionPlan getExecutionPlan();
-
-    static PolicyExecutionPlan parseExecutionPlan(String executionPlan) throws Exception {
-        return new PolicyExecutionPlannerImpl(executionPlan).getExecutionPlan();
-    }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/eagle/blob/6fd95d5c/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/interpreter/PolicyExecutionPlannerImpl.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/interpreter/PolicyExecutionPlannerImpl.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/interpreter/PolicyExecutionPlannerImpl.java
deleted file mode 100644
index 4e6901d..0000000
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/interpreter/PolicyExecutionPlannerImpl.java
+++ /dev/null
@@ -1,376 +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.alert.engine.interpreter;
-
-import com.google.common.base.Preconditions;
-import org.apache.commons.collections.ListUtils;
-import org.apache.eagle.alert.engine.coordinator.StreamColumn;
-import org.apache.eagle.alert.engine.coordinator.StreamPartition;
-import org.apache.eagle.alert.engine.coordinator.StreamSortSpec;
-import org.apache.eagle.alert.engine.evaluator.impl.SiddhiDefinitionAdapter;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.wso2.siddhi.core.exception.DefinitionNotExistException;
-import org.wso2.siddhi.query.api.ExecutionPlan;
-import org.wso2.siddhi.query.api.exception.DuplicateDefinitionException;
-import org.wso2.siddhi.query.api.exception.ExecutionPlanValidationException;
-import org.wso2.siddhi.query.api.execution.ExecutionElement;
-import org.wso2.siddhi.query.api.execution.query.Query;
-import org.wso2.siddhi.query.api.execution.query.input.handler.StreamHandler;
-import org.wso2.siddhi.query.api.execution.query.input.handler.Window;
-import org.wso2.siddhi.query.api.execution.query.input.state.*;
-import org.wso2.siddhi.query.api.execution.query.input.stream.*;
-import org.wso2.siddhi.query.api.execution.query.output.stream.OutputStream;
-import org.wso2.siddhi.query.api.execution.query.selection.OutputAttribute;
-import org.wso2.siddhi.query.api.execution.query.selection.Selector;
-import org.wso2.siddhi.query.api.expression.Expression;
-import org.wso2.siddhi.query.api.expression.Variable;
-import org.wso2.siddhi.query.api.expression.condition.Compare;
-import org.wso2.siddhi.query.api.expression.constant.IntConstant;
-import org.wso2.siddhi.query.api.expression.constant.LongConstant;
-import org.wso2.siddhi.query.api.expression.constant.TimeConstant;
-import org.wso2.siddhi.query.compiler.SiddhiCompiler;
-
-import java.util.*;
-import java.util.stream.Collectors;
-
-class PolicyExecutionPlannerImpl implements PolicyExecutionPlanner {
-
-    private static final Logger LOG = LoggerFactory.getLogger(PolicyExecutionPlannerImpl.class);
-
-    /**
-     * See https://docs.wso2.com/display/CEP300/Windows#Windows-ExternalTimeWindow.
-     */
-    private static final String WINDOW_EXTERNAL_TIME = "externalTime";
-
-    private final String executionPlan;
-    private final Map<String,List<StreamColumn>> effectiveInputStreams;
-    private final Map<String, String> effectiveInputStreamsAlias;
-    private final Map<String,List<StreamColumn>> effectiveOutputStreams;
-    private final Map<String,StreamPartition> effectivePartitions;
-    private final PolicyExecutionPlan policyExecutionPlan;
-
-    public PolicyExecutionPlannerImpl(String executionPlan) throws Exception {
-        this.executionPlan = executionPlan;
-        this.effectiveInputStreams = new HashMap<>();
-        this.effectiveInputStreamsAlias = new HashMap<>();
-        this.effectiveOutputStreams = new HashMap<>();
-        this.effectivePartitions = new HashMap<>();
-        this.policyExecutionPlan = doParse();
-    }
-
-    @Override
-    public PolicyExecutionPlan getExecutionPlan() {
-        return policyExecutionPlan;
-    }
-
-    private PolicyExecutionPlan doParse()  throws Exception {
-        PolicyExecutionPlan policyExecutionPlan = new PolicyExecutionPlan();
-        try {
-            ExecutionPlan executionPlan = SiddhiCompiler.parse(this.executionPlan);
-
-            policyExecutionPlan.setExecutionPlanDesc(executionPlan.toString());
-
-            // Set current execution plan as valid
-            policyExecutionPlan.setExecutionPlanSource(this.executionPlan);
-            policyExecutionPlan.setInternalExecutionPlan(executionPlan);
-
-
-            // Go through execution element
-            for (ExecutionElement executionElement : executionPlan.getExecutionElementList()) {
-                // -------------
-                // Explain Query
-                // -------------
-                if (executionElement instanceof Query) {
-                    // -----------------------
-                    // Query Level Variables
-                    // -----------------------
-                    InputStream inputStream = ((Query) executionElement).getInputStream();
-                    Selector selector = ((Query) executionElement).getSelector();
-                    Map<String, SingleInputStream> queryLevelAliasToStreamMapping = new HashMap<>();
-
-                    // Inputs stream definitions
-                    for (String streamId : inputStream.getUniqueStreamIds()) {
-                        if (!effectiveInputStreams.containsKey(streamId)) {
-                            org.wso2.siddhi.query.api.definition.StreamDefinition streamDefinition = executionPlan.getStreamDefinitionMap().get(streamId);
-                            if (streamDefinition != null) {
-                                effectiveInputStreams.put(streamId, SiddhiDefinitionAdapter.convertFromSiddiDefinition(streamDefinition).getColumns());
-                            } else {
-                                effectiveInputStreams.put(streamId, null);
-                            }
-                        }
-                    }
-
-                    // Window Spec and Partition
-                    if (inputStream instanceof SingleInputStream) {
-                        retrieveAliasForQuery((SingleInputStream) inputStream, queryLevelAliasToStreamMapping);
-                        retrievePartition(findStreamPartition((SingleInputStream) inputStream, selector));
-                    } else {
-                        if (inputStream instanceof JoinInputStream) {
-                            // Only Support JOIN/INNER_JOIN Now
-                            if (((JoinInputStream) inputStream).getType().equals(JoinInputStream.Type.INNER_JOIN) || ((JoinInputStream) inputStream).getType().equals(JoinInputStream.Type.JOIN)) {
-                                SingleInputStream leftInputStream = (SingleInputStream) ((JoinInputStream) inputStream).getLeftInputStream();
-                                SingleInputStream rightInputStream = (SingleInputStream) ((JoinInputStream) inputStream).getRightInputStream();
-
-                                retrievePartition(findStreamPartition(leftInputStream, selector));
-                                retrievePartition(findStreamPartition(rightInputStream, selector));
-                                retrieveAliasForQuery(leftInputStream, queryLevelAliasToStreamMapping);
-                                retrieveAliasForQuery(rightInputStream, queryLevelAliasToStreamMapping);
-
-                            } else {
-                                throw new ExecutionPlanValidationException("Not support " + ((JoinInputStream) inputStream).getType() + " yet, currently support: INNER JOIN");
-                            }
-
-                            Expression joinCondition = ((JoinInputStream) inputStream).getOnCompare();
-
-                            if (joinCondition != null) {
-                                if (joinCondition instanceof Compare) {
-                                    if (((Compare) joinCondition).getOperator().equals(Compare.Operator.EQUAL)) {
-                                        Variable leftExpression = (Variable) ((Compare) joinCondition).getLeftExpression();
-                                        Preconditions.checkNotNull(leftExpression.getStreamId());
-                                        Preconditions.checkNotNull(leftExpression.getAttributeName());
-
-                                        StreamPartition leftPartition = new StreamPartition();
-                                        leftPartition.setType(StreamPartition.Type.GROUPBY);
-                                        leftPartition.setColumns(Collections.singletonList(leftExpression.getAttributeName()));
-                                        leftPartition.setStreamId(retrieveStreamId(leftExpression, effectiveInputStreams,queryLevelAliasToStreamMapping));
-                                        retrievePartition(leftPartition);
-
-                                        Variable rightExpression = (Variable) ((Compare) joinCondition).getRightExpression();
-                                        Preconditions.checkNotNull(rightExpression.getStreamId());
-                                        Preconditions.checkNotNull(rightExpression.getAttributeName());
-                                        StreamPartition rightPartition = new StreamPartition();
-                                        rightPartition.setType(StreamPartition.Type.GROUPBY);
-                                        rightPartition.setColumns(Collections.singletonList(rightExpression.getAttributeName()));
-                                        rightPartition.setStreamId(retrieveStreamId(rightExpression, effectiveInputStreams,queryLevelAliasToStreamMapping));
-                                        retrievePartition(leftPartition);
-                                    } else {
-                                        throw new ExecutionPlanValidationException("Only support \"EQUAL\" condition in INNER JOIN" + joinCondition);
-                                    }
-                                } else {
-                                    throw new ExecutionPlanValidationException("Only support \"Compare\" on INNER JOIN condition in INNER JOIN: " + joinCondition);
-                                }
-                            }
-                        } else if (inputStream instanceof StateInputStream) {
-                            // Group By Spec
-                            List<Variable> groupBy = selector.getGroupByList();
-                            if (groupBy.size() >= 0) {
-                                Map<String, List<Variable>> streamGroupBy = new HashMap<>();
-                                for (String streamId : inputStream.getUniqueStreamIds()) {
-                                    streamGroupBy.put(streamId, new ArrayList<>());
-                                }
-
-                                collectStreamReferenceIdMapping(((StateInputStream)inputStream).getStateElement());
-
-                                for (Variable variable : groupBy) {
-                                    // Not stream not set, then should be all streams' same field
-                                    if (variable.getStreamId() == null) {
-                                        for (String streamId : inputStream.getUniqueStreamIds()) {
-                                            streamGroupBy.get(streamId).add(variable);
-                                        }
-                                    } else {
-                                        String streamId = variable.getStreamId();
-                                        if (!this.effectiveInputStreamsAlias.containsKey(streamId)) {
-                                            streamId = retrieveStreamId(variable, effectiveInputStreams,queryLevelAliasToStreamMapping);
-                                        } else {
-                                            streamId = this.effectiveInputStreamsAlias.get(streamId);
-                                        }
-                                        if (streamGroupBy.containsKey(streamId)) {
-                                            streamGroupBy.get(streamId).add(variable);
-                                        } else {
-                                            throw new DefinitionNotExistException(streamId);
-                                        }
-                                    }
-                                }
-                                for (Map.Entry<String, List<Variable>> entry : streamGroupBy.entrySet()) {
-                                    if (entry.getValue().size() > 0) {
-                                        StreamPartition partition = generatePartition(entry.getKey(), null, Arrays.asList(entry.getValue().toArray(new Variable[entry.getValue().size()])));
-                                        if (((StateInputStream) inputStream).getStateType().equals(StateInputStream.Type.PATTERN)
-                                                || ((StateInputStream) inputStream).getStateType().equals(StateInputStream.Type.SEQUENCE)) {
-                                            if (effectivePartitions.containsKey(partition.getStreamId())) {
-                                                StreamPartition existingPartition = effectivePartitions.get(partition.getStreamId());
-                                                if (!existingPartition.equals(partition)
-                                                        && existingPartition.getType().equals(partition.getType())
-                                                        && ListUtils.isEqualList(existingPartition.getColumns(), partition.getColumns())) {
-                                                    partition.setSortSpec(existingPartition.getSortSpec());
-                                                }
-                                            }
-                                        }
-                                        retrievePartition(partition);
-                                    }
-                                }
-                            }
-                        }
-                    }
-
-                    // Output streams
-                    OutputStream outputStream = ((Query) executionElement).getOutputStream();
-                    effectiveOutputStreams.put(outputStream.getId(), convertOutputStreamColumns(selector.getSelectionList()));
-                } else {
-                    LOG.warn("Unhandled execution element: {}", executionElement.toString());
-                }
-            }
-            // Set effective input streams
-            policyExecutionPlan.setInputStreams(effectiveInputStreams);
-
-            // Set effective output streams
-            policyExecutionPlan.setOutputStreams(effectiveOutputStreams);
-
-            // Set Partitions
-            for (String streamId : effectiveInputStreams.keySet()) {
-                // Use shuffle partition by default
-                if (!effectivePartitions.containsKey(streamId)) {
-                    StreamPartition shufflePartition = new StreamPartition();
-                    shufflePartition.setStreamId(streamId);
-                    shufflePartition.setType(StreamPartition.Type.SHUFFLE);
-                    effectivePartitions.put(streamId, shufflePartition);
-                }
-            }
-            policyExecutionPlan.setStreamPartitions(new ArrayList<>(effectivePartitions.values()));
-        } catch (Exception ex) {
-            LOG.error("Got error to parse policy execution plan: \n{}", this.executionPlan, ex);
-            throw ex;
-        }
-        return policyExecutionPlan;
-    }
-
-    private void collectStreamReferenceIdMapping(StateElement stateElement) {
-        if (stateElement instanceof LogicalStateElement) {
-            collectStreamReferenceIdMapping(((LogicalStateElement) stateElement).getStreamStateElement1());
-            collectStreamReferenceIdMapping(((LogicalStateElement) stateElement).getStreamStateElement2());
-        } else if (stateElement instanceof CountStateElement) {
-            collectStreamReferenceIdMapping(((CountStateElement) stateElement).getStreamStateElement());
-        } else if (stateElement instanceof EveryStateElement) {
-            collectStreamReferenceIdMapping(((EveryStateElement) stateElement).getStateElement());
-        } else if (stateElement instanceof NextStateElement) {
-            collectStreamReferenceIdMapping(((NextStateElement) stateElement).getStateElement());
-            collectStreamReferenceIdMapping(((NextStateElement) stateElement).getNextStateElement());
-        } else if (stateElement instanceof StreamStateElement) {
-            BasicSingleInputStream basicSingleInputStream = ((StreamStateElement) stateElement).getBasicSingleInputStream();
-            this.effectiveInputStreamsAlias.put(basicSingleInputStream.getStreamReferenceId(), basicSingleInputStream.getStreamId());
-        }
-    }
-
-    private String retrieveStreamId(Variable variable, Map<String, List<StreamColumn>> streamMap, Map<String, SingleInputStream> aliasMap) {
-        Preconditions.checkNotNull(variable.getStreamId(), "streamId");
-        if (streamMap.containsKey(variable.getStreamId()) && aliasMap.containsKey(variable.getStreamId())) {
-            throw new DuplicateDefinitionException("Duplicated streamId and alias: " + variable.getStreamId());
-        } else if (streamMap.containsKey(variable.getStreamId())) {
-            return variable.getStreamId();
-        } else if (aliasMap.containsKey(variable.getStreamId())) {
-            return aliasMap.get(variable.getStreamId()).getStreamId();
-        } else {
-            throw new DefinitionNotExistException(variable.getStreamId());
-        }
-    }
-
-    private StreamPartition findStreamPartition(SingleInputStream inputStream, Selector selector) {
-        // Window Spec
-        List<Window> windows = new ArrayList<>();
-        for (StreamHandler streamHandler : inputStream.getStreamHandlers()) {
-            if (streamHandler instanceof Window) {
-                windows.add((Window) streamHandler);
-            }
-        }
-
-        // Group By Spec
-        List<Variable> groupBy = selector.getGroupByList();
-        if (windows.size() > 0 || groupBy.size() >= 0) {
-            return generatePartition(inputStream.getStreamId(), windows, groupBy);
-        } else {
-            return null;
-        }
-    }
-
-    private void retrievePartition(StreamPartition partition) {
-        if (partition == null) {
-            return;
-        }
-
-        if (!effectivePartitions.containsKey(partition.getStreamId())) {
-            effectivePartitions.put(partition.getStreamId(), partition);
-        } else if (!effectivePartitions.get(partition.getStreamId()).equals(partition)) {
-            StreamPartition existingPartition = effectivePartitions.get(partition.getStreamId());
-            // If same Type & Columns but different sort spec, then use larger
-            if (existingPartition.getType().equals(partition.getType())
-                && ListUtils.isEqualList(existingPartition.getColumns(), partition.getColumns())
-                && partition.getSortSpec().getWindowPeriodMillis() > existingPartition.getSortSpec().getWindowPeriodMillis()
-                || existingPartition.getType().equals(StreamPartition.Type.SHUFFLE)) {
-                effectivePartitions.put(partition.getStreamId(), partition);
-            } else {
-                // Throw exception as it unable to conflict effectivePartitions on same stream will not be able to run in distributed mode
-                throw new ExecutionPlanValidationException("You have incompatible partitions on stream " + partition.getStreamId()
-                    + ": [1] " + effectivePartitions.get(partition.getStreamId()).toString() + " [2] " + partition.toString() + "");
-            }
-        }
-    }
-
-    private void retrieveAliasForQuery(SingleInputStream inputStream, Map<String, SingleInputStream> aliasStreamMapping) {
-        if (inputStream.getStreamReferenceId() != null) {
-            if (aliasStreamMapping.containsKey(inputStream.getStreamReferenceId())) {
-                throw new ExecutionPlanValidationException("Duplicated stream alias " + inputStream.getStreamId() + " -> " + inputStream);
-            } else {
-                aliasStreamMapping.put(inputStream.getStreamReferenceId(), inputStream);
-            }
-        }
-    }
-
-    private StreamPartition generatePartition(String streamId, List<Window> windows, List<Variable> groupBy) {
-        StreamPartition partition = new StreamPartition();
-        partition.setStreamId(streamId);
-        StreamSortSpec sortSpec = null;
-        if (windows != null && windows.size() > 0) {
-            for (Window window : windows) {
-                if (window.getFunction().equals(WINDOW_EXTERNAL_TIME)) {
-                    sortSpec = new StreamSortSpec();
-                    sortSpec.setWindowPeriodMillis(getExternalTimeWindowSize(window));
-                    sortSpec.setWindowMargin(sortSpec.getWindowPeriodMillis() / 5);
-                }
-            }
-        }
-        partition.setSortSpec(sortSpec);
-        if (groupBy != null && groupBy.size() > 0) {
-            partition.setColumns(groupBy.stream().map(Variable::getAttributeName).collect(Collectors.toList()));
-            partition.setType(StreamPartition.Type.GROUPBY);
-        } else {
-            partition.setType(StreamPartition.Type.SHUFFLE);
-        }
-        return partition;
-    }
-
-    private static int getExternalTimeWindowSize(Window window) {
-        Expression windowSize = window.getParameters()[1];
-        if (windowSize instanceof TimeConstant) {
-            return ((TimeConstant) windowSize).getValue().intValue();
-        } else if (windowSize instanceof IntConstant) {
-            return ((IntConstant) windowSize).getValue();
-        } else if (windowSize instanceof LongConstant) {
-            return ((LongConstant) windowSize).getValue().intValue();
-        } else {
-            throw new UnsupportedOperationException("Illegal type of window size expression:" + windowSize.toString());
-        }
-    }
-
-    private static List<StreamColumn> convertOutputStreamColumns(List<OutputAttribute> outputAttributeList) {
-        return outputAttributeList.stream().map(outputAttribute -> {
-            StreamColumn streamColumn = new StreamColumn();
-            streamColumn.setName(outputAttribute.getRename());
-            streamColumn.setDescription(outputAttribute.getExpression().toString());
-            return streamColumn;
-        }).collect(Collectors.toList());
-    }
-}

http://git-wip-us.apache.org/repos/asf/eagle/blob/6fd95d5c/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/interpreter/PolicyInterpreter.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/interpreter/PolicyInterpreter.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/interpreter/PolicyInterpreter.java
deleted file mode 100644
index 4add3ff..0000000
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/interpreter/PolicyInterpreter.java
+++ /dev/null
@@ -1,116 +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.alert.engine.interpreter;
-
-import com.google.common.base.Preconditions;
-import org.apache.eagle.alert.engine.coordinator.PolicyDefinition;
-import org.apache.eagle.alert.engine.coordinator.StreamDefinition;
-import org.apache.eagle.alert.engine.coordinator.StreamNotDefinedException;
-import org.apache.eagle.alert.engine.evaluator.PolicyStreamHandlers;
-import org.apache.eagle.alert.engine.evaluator.impl.SiddhiDefinitionAdapter;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.HashMap;
-import java.util.Map;
-
-/**
- * PolicyInterpreter Helper Methods:
- * <ul>
- * <li>Parse: parse siddhi query and generate static execution plan</li>
- * <li>Validate: validate policy definition with execution plan and metadata</li>
- * </ul>
- *
- * @see PolicyExecutionPlanner
- * @see <a href="https://docs.wso2.com/display/CEP300/WSO2+Complex+Event+Processor+Documentation">WSO2 Complex Event Processor Documentation</a>
- */
-public class PolicyInterpreter {
-    private static final Logger LOG = LoggerFactory.getLogger(PolicyInterpreter.class);
-
-    /**
-     * See https://docs.wso2.com/display/CEP300/Windows#Windows-ExternalTimeWindow.
-     */
-    private static final String WINDOW_EXTERNAL_TIME = "externalTime";
-
-    public static PolicyParseResult parse(String executionPlanQuery) {
-        PolicyParseResult policyParseResult = new PolicyParseResult();
-        try {
-            policyParseResult.setPolicyExecutionPlan(parseExecutionPlan(executionPlanQuery));
-            policyParseResult.setSuccess(true);
-            policyParseResult.setMessage("Parsed successfully");
-        } catch (Exception exception) {
-            LOG.error("Got error to parse policy: {}", executionPlanQuery, exception);
-            policyParseResult.setSuccess(false);
-            policyParseResult.setMessage(exception.getMessage());
-            policyParseResult.setStackTrace(exception);
-        }
-        return policyParseResult;
-    }
-
-    /**
-     * Quick parseExecutionPlan policy.
-     */
-    public static PolicyExecutionPlan parseExecutionPlan(String policyDefinition, Map<String, StreamDefinition> inputStreamDefinitions) throws Exception {
-        // Validate inputStreams are valid
-        Preconditions.checkNotNull(inputStreamDefinitions, "No inputStreams to connect from");
-        return parseExecutionPlan(SiddhiDefinitionAdapter.buildSiddhiExecutionPlan(policyDefinition, inputStreamDefinitions));
-    }
-
-    public static PolicyExecutionPlan parseExecutionPlan(String executionPlanQuery) throws Exception {
-        return PolicyExecutionPlanner.parseExecutionPlan(executionPlanQuery);
-    }
-
-    public static PolicyValidationResult validate(PolicyDefinition policy, Map<String, StreamDefinition> allDefinitions) {
-        Map<String, StreamDefinition> inputDefinitions = new HashMap<>();
-        PolicyValidationResult policyValidationResult = new PolicyValidationResult();
-        policyValidationResult.setPolicyDefinition(policy);
-        try {
-            if (policy.getInputStreams() != null) {
-                for (String streamId : policy.getInputStreams()) {
-                    if (allDefinitions.containsKey(streamId)) {
-                        inputDefinitions.put(streamId, allDefinitions.get(streamId));
-                    } else {
-                        throw new StreamNotDefinedException(streamId);
-                    }
-                }
-            }
-
-            PolicyExecutionPlan policyExecutionPlan = null;
-            if (PolicyStreamHandlers.SIDDHI_ENGINE.equalsIgnoreCase(policy.getDefinition().getType())) {
-                policyExecutionPlan = parseExecutionPlan(policy.getDefinition().getValue(), inputDefinitions);
-                // Validate output
-                if (policy.getOutputStreams() != null) {
-                    for (String outputStream : policy.getOutputStreams()) {
-                        if (!policyExecutionPlan.getOutputStreams().containsKey(outputStream)) {
-                            throw new StreamNotDefinedException("Output stream " + outputStream + " not defined");
-                        }
-                    }
-                }
-            }
-            policyValidationResult.setPolicyExecutionPlan(policyExecutionPlan);
-            policyValidationResult.setSuccess(true);
-            policyValidationResult.setMessage("Validated successfully");
-        } catch (Exception exception) {
-            LOG.error("Got error to validate policy definition: {}", policy, exception);
-            policyValidationResult.setSuccess(false);
-            policyValidationResult.setMessage(exception.getMessage());
-            policyValidationResult.setStackTrace(exception);
-        }
-
-        return policyValidationResult;
-    }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/eagle/blob/6fd95d5c/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/interpreter/PolicyParseResult.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/interpreter/PolicyParseResult.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/interpreter/PolicyParseResult.java
deleted file mode 100644
index a0f3ad2..0000000
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/interpreter/PolicyParseResult.java
+++ /dev/null
@@ -1,65 +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.alert.engine.interpreter;
-
-import com.fasterxml.jackson.databind.annotation.JsonSerialize;
-import org.apache.commons.lang3.exception.ExceptionUtils;
-
-@JsonSerialize(include = JsonSerialize.Inclusion.NON_NULL)
-public class PolicyParseResult {
-    private boolean success;
-    private String message;
-    private String exception;
-
-    private PolicyExecutionPlan policyExecutionPlan;
-
-    public String getException() {
-        return exception;
-    }
-
-    public void setException(String exception) {
-        this.exception = exception;
-    }
-
-    public String getMessage() {
-        return message;
-    }
-
-    public void setMessage(String message) {
-        this.message = message;
-    }
-
-    public void setStackTrace(Throwable throwable) {
-        this.setException(ExceptionUtils.getStackTrace(throwable));
-    }
-
-    public boolean isSuccess() {
-        return success;
-    }
-
-    public void setSuccess(boolean success) {
-        this.success = success;
-    }
-
-    public PolicyExecutionPlan getPolicyExecutionPlan() {
-        return policyExecutionPlan;
-    }
-
-    public void setPolicyExecutionPlan(PolicyExecutionPlan policyExecutionPlan) {
-        this.policyExecutionPlan = policyExecutionPlan;
-    }
-}

http://git-wip-us.apache.org/repos/asf/eagle/blob/6fd95d5c/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/interpreter/PolicyValidationResult.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/interpreter/PolicyValidationResult.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/interpreter/PolicyValidationResult.java
deleted file mode 100644
index 17f6091..0000000
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/interpreter/PolicyValidationResult.java
+++ /dev/null
@@ -1,76 +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.alert.engine.interpreter;
-
-
-import com.fasterxml.jackson.databind.annotation.JsonSerialize;
-import org.apache.commons.lang3.exception.ExceptionUtils;
-import org.apache.eagle.alert.engine.coordinator.PolicyDefinition;
-
-@JsonSerialize(include = JsonSerialize.Inclusion.NON_NULL)
-public class PolicyValidationResult {
-    private boolean success;
-    private String message;
-    private String exception;
-
-    private PolicyExecutionPlan policyExecutionPlan;
-    private PolicyDefinition policyDefinition;
-
-    public String getException() {
-        return exception;
-    }
-
-    public void setException(String exception) {
-        this.exception = exception;
-    }
-
-    public String getMessage() {
-        return message;
-    }
-
-    public void setMessage(String message) {
-        this.message = message;
-    }
-
-    public void setStackTrace(Throwable throwable) {
-        this.setException(ExceptionUtils.getStackTrace(throwable));
-    }
-
-    public boolean isSuccess() {
-        return success;
-    }
-
-    public void setSuccess(boolean success) {
-        this.success = success;
-    }
-
-    public PolicyExecutionPlan getPolicyExecutionPlan() {
-        return policyExecutionPlan;
-    }
-
-    public void setPolicyExecutionPlan(PolicyExecutionPlan policyExecutionPlan) {
-        this.policyExecutionPlan = policyExecutionPlan;
-    }
-
-    public PolicyDefinition getPolicyDefinition() {
-        return policyDefinition;
-    }
-
-    public void setPolicyDefinition(PolicyDefinition policyDefinition) {
-        this.policyDefinition = policyDefinition;
-    }
-}
\ No newline at end of file


Mime
View raw message