eagle-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From h..@apache.org
Subject [22/46] incubator-eagle git commit: [EAGLE-325] Initialize next-gen alert engine code on branch-0.5
Date Thu, 02 Jun 2016 07:08:01 GMT
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/coordination/model/internal/Topology.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/coordination/model/internal/Topology.java b/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/coordination/model/internal/Topology.java
deleted file mode 100644
index 189e2a5..0000000
--- a/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/coordination/model/internal/Topology.java
+++ /dev/null
@@ -1,170 +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.coordination.model.internal;
-
-import java.util.HashSet;
-import java.util.Set;
-
-/**
- * @since Mar 24, 2016 Logically one unit topology consists of S spouts, G
- *        groupby bolts, A alertBolts normally S=1 Physically each spout is
- *        composed of s spout nodes, each groupby bolt is composed of g groupby
- *        nodes, and each alert bolt is composed of a alert nodes
- */
-public class Topology {
-
-    private String name;
-    // number of logical nodes
-    private int numOfSpout;
-    private int numOfAlertBolt;
-    private int numOfGroupBolt;
-    private int numOfPublishBolt;
-    private String spoutId;
-    private String pubBoltId;
-    @Deprecated
-    private Set<String> groupNodeIds;
-    @Deprecated
-    private Set<String> alertBoltIds;
-
-    // number of physical nodes for each logic bolt
-    private int spoutParallelism = 1;
-    private int groupParallelism = 1;
-    private int alertParallelism = 1;
-
-    private String clusterName;
-
-    public Topology() {
-    }
-
-    public Topology(String name, int group, int alert) {
-        this.name = name;
-        this.numOfSpout = 1;
-        this.numOfGroupBolt = group;
-        this.numOfAlertBolt = alert;
-        groupNodeIds = new HashSet<String>(group);
-        alertBoltIds = new HashSet<String>(alert);
-
-        spoutParallelism = 1;
-        groupParallelism = 1;
-        alertParallelism = 1;
-    }
-
-    public String getName() {
-        return name;
-    }
-
-    public void setName(String name) {
-        this.name = name;
-    }
-
-    public int getNumOfSpout() {
-        return numOfSpout;
-    }
-
-    public void setNumOfSpout(int numOfSpout) {
-        this.numOfSpout = numOfSpout;
-    }
-
-    public int getNumOfAlertBolt() {
-        return numOfAlertBolt;
-    }
-
-    public void setNumOfAlertBolt(int numOfAlertBolt) {
-        this.numOfAlertBolt = numOfAlertBolt;
-    }
-
-    public int getNumOfGroupBolt() {
-        return numOfGroupBolt;
-    }
-
-    public void setNumOfGroupBolt(int numOfGroupBolt) {
-        this.numOfGroupBolt = numOfGroupBolt;
-    }
-
-    public String getSpoutId() {
-        return spoutId;
-    }
-
-    public void setSpoutId(String spoutId) {
-        this.spoutId = spoutId;
-    }
-
-    public String getPubBoltId() {
-        return pubBoltId;
-    }
-
-    public void setPubBoltId(String pubBoltId) {
-        this.pubBoltId = pubBoltId;
-    }
-
-    public Set<String> getGroupNodeIds() {
-        return groupNodeIds;
-    }
-
-    public void setGroupNodeIds(Set<String> groupNodeIds) {
-        this.groupNodeIds = groupNodeIds;
-    }
-
-    public Set<String> getAlertBoltIds() {
-        return alertBoltIds;
-    }
-
-    public void setAlertBoltIds(Set<String> alertBoltIds) {
-        this.alertBoltIds = alertBoltIds;
-    }
-
-    public int getNumOfPublishBolt() {
-        return numOfPublishBolt;
-    }
-
-    public void setNumOfPublishBolt(int numOfPublishBolt) {
-        this.numOfPublishBolt = numOfPublishBolt;
-    }
-
-    public int getSpoutParallelism() {
-        return spoutParallelism;
-    }
-
-    public void setSpoutParallelism(int spoutParallelism) {
-        this.spoutParallelism = spoutParallelism;
-    }
-
-    public int getGroupParallelism() {
-        return groupParallelism;
-    }
-
-    public void setGroupParallelism(int groupParallelism) {
-        this.groupParallelism = groupParallelism;
-    }
-
-    public int getAlertParallelism() {
-        return alertParallelism;
-    }
-
-    public void setAlertParallelism(int alertParallelism) {
-        this.alertParallelism = alertParallelism;
-    }
-
-    public String getClusterName() {
-        return clusterName;
-    }
-
-    public void setClusterName(String clusterName) {
-        this.clusterName = clusterName;
-    }
-
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/codec/IStreamCodec.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/codec/IStreamCodec.java b/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/codec/IStreamCodec.java
deleted file mode 100644
index 3272d28..0000000
--- a/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/codec/IStreamCodec.java
+++ /dev/null
@@ -1,30 +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.codec;
-
-import org.apache.eagle.alert.engine.model.StreamEvent;
-
-/**
- * @since Apr 5, 2016
- *
- */
-public interface IStreamCodec {
-
-    StreamEvent decode(byte[] contents);
-
-    byte[] encode(StreamEvent tuple);
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/codec/SherlockEventCodec.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/codec/SherlockEventCodec.java b/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/codec/SherlockEventCodec.java
deleted file mode 100644
index 49726e9..0000000
--- a/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/codec/SherlockEventCodec.java
+++ /dev/null
@@ -1,41 +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.codec;
-
-import org.apache.eagle.alert.engine.model.StreamEvent;
-
-/**
- * This is used for event codec.
- * 
- * @since Apr 5, 2016
- *
- */
-public class SherlockEventCodec implements IStreamCodec {
-
-    @Override
-    public StreamEvent decode(byte[] contents) {
-        // TODO Auto-generated method stub
-        return null;
-    }
-
-    @Override
-    public byte[] encode(StreamEvent tuple) {
-        // TODO Auto-generated method stub
-        return null;
-    }
-
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/codec/SherlockMetricCodec.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/codec/SherlockMetricCodec.java b/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/codec/SherlockMetricCodec.java
deleted file mode 100644
index 0e875d8..0000000
--- a/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/codec/SherlockMetricCodec.java
+++ /dev/null
@@ -1,39 +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.codec;
-
-import org.apache.eagle.alert.engine.model.StreamEvent;
-
-/**
- * @since Apr 5, 2016
- *
- */
-public class SherlockMetricCodec implements IStreamCodec {
-
-    @Override
-    public StreamEvent decode(byte[] contents) {
-        // TODO Auto-generated method stub
-        return null;
-    }
-
-    @Override
-    public byte[] encode(StreamEvent tuple) {
-        // TODO Auto-generated method stub
-        return null;
-    }
-
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/coordinator/PolicyDefinition.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/coordinator/PolicyDefinition.java b/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/coordinator/PolicyDefinition.java
deleted file mode 100644
index f97eb2b..0000000
--- a/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/coordinator/PolicyDefinition.java
+++ /dev/null
@@ -1,199 +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.coordinator;
-
-import java.io.Serializable;
-import java.util.ArrayList;
-import java.util.List;
-
-import org.apache.commons.collections.CollectionUtils;
-import org.apache.commons.lang3.builder.HashCodeBuilder;
-
-/**
- * @since Apr 5, 2016
- *
- */
-public class PolicyDefinition implements Serializable{
-    private static final long serialVersionUID = 377581499339572414L;
-    // unique identifier
-    private String name;
-    private String description;
-    private List<String> inputStreams = new ArrayList<String>();
-    private List<String> outputStreams = new ArrayList<String>();
-
-    private Definition definition;
-
-    // one stream only have one partition in one policy, since we don't support stream alias
-    private List<StreamPartition> partitionSpec = new ArrayList<StreamPartition>();
-
-    // runtime configuration for policy, these are user-invisible
-    private int parallelismHint = 1;
-
-    public String getName() {
-        return name;
-    }
-
-    public void setName(String name) {
-        this.name = name;
-    }
-
-    public String getDescription() {
-        return description;
-    }
-
-    public void setDescription(String description) {
-        this.description = description;
-    }
-
-    public List<String> getInputStreams() {
-        return inputStreams;
-    }
-
-    public void setInputStreams(List<String> inputStreams) {
-        this.inputStreams = inputStreams;
-    }
-
-    public List<String> getOutputStreams() {
-        return outputStreams;
-    }
-
-    public void setOutputStreams(List<String> outputStreams) {
-        this.outputStreams = outputStreams;
-    }
-
-    public Definition getDefinition() {
-        return definition;
-    }
-
-    public void setDefinition(Definition definition) {
-        this.definition = definition;
-    }
-
-    public List<StreamPartition> getPartitionSpec() {
-        return partitionSpec;
-    }
-
-    public void setPartitionSpec(List<StreamPartition> partitionSpec) {
-        this.partitionSpec = partitionSpec;
-    }
-
-    public void addPartition(StreamPartition par) {
-        this.partitionSpec.add(par);
-    }
-
-    public int getParallelismHint() {
-        return parallelismHint;
-    }
-
-    public void setParallelismHint(int parallelism) {
-        this.parallelismHint = parallelism;
-    }
-
-    @Override
-    public int hashCode() {
-        return new HashCodeBuilder().
-                append(name).
-                append(description).
-                append(inputStreams).
-                append(outputStreams).
-                append(definition).
-                append(partitionSpec).
-//                append(parallelismHint).
-                build();
-    }
-
-    @Override
-    public boolean equals(Object that){
-        if(that == this)
-            return true;
-        if(! (that instanceof PolicyDefinition))
-            return false;
-        PolicyDefinition another = (PolicyDefinition)that;
-        if(another.name.equals(this.name) &&
-                another.description.equals(this.description) &&
-                CollectionUtils.isEqualCollection(another.inputStreams, this.inputStreams) &&
-                CollectionUtils.isEqualCollection(another.outputStreams, this.outputStreams) &&
-                another.definition.equals(this.definition) &&
-                CollectionUtils.isEqualCollection(another.partitionSpec, this.partitionSpec) 
-//                && another.parallelismHint == this.parallelismHint
-                ) {
-            return true;
-        }
-        return false;
-    }
-
-    public static class Definition implements Serializable{
-        private static final long serialVersionUID = -622366527887848346L;
-
-        public String type;
-        public String value;
-
-        public Definition(String type,String value){
-            this.type = type;
-            this.value = value;
-        }
-
-        public Definition() {
-            this.type = null;
-            this.value = null;
-        }
-
-        @Override
-        public int hashCode() {
-            return new HashCodeBuilder().append(type).append(value).build();
-        }
-
-        @Override
-        public boolean equals(Object that){
-            if(that == this)
-                return true;
-            if(!(that instanceof Definition))
-                return false;
-            Definition another = (Definition)that;
-            if(another.type.equals(this.type) &&
-                    another.value.equals(this.value))
-                return true;
-            return false;
-        }
-
-        public String getType() {
-            return type;
-        }
-
-        public void setType(String type) {
-            this.type = type;
-        }
-
-        public String getValue() {
-            return value;
-        }
-
-        public void setValue(String value) {
-            this.value = value;
-        }
-
-        @Override
-        public String toString() {
-            return String.format("{type=\"%s\",value=\"%s\"",type,value);
-        }
-    }
-
-    @Override
-    public String toString() {
-        return String.format("{name=\"%s\",definition=%s}",this.getName(),this.getDefinition().toString());
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/coordinator/Publishment.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/coordinator/Publishment.java b/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/coordinator/Publishment.java
deleted file mode 100644
index d8b4f28..0000000
--- a/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/coordinator/Publishment.java
+++ /dev/null
@@ -1,101 +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.coordinator;
-
-import org.apache.commons.lang3.builder.HashCodeBuilder;
-
-import java.util.List;
-import java.util.Map;
-import java.util.Objects;
-
-/**
- * @since Apr 11, 2016
- *
- */
-public class Publishment {
-
-    private String name;
-    private String type;
-    private List<String> policyIds;
-    private String dedupIntervalMin;
-    private Map<String, String> properties;
-
-    public String getName() {
-        return name;
-    }
-
-    public void setName(String name) {
-        this.name = name;
-    }
-
-    public String getType() {
-        return type;
-    }
-
-    public void setType(String type) {
-        this.type = type;
-    }
-
-    public List<String> getPolicyIds() {
-        return policyIds;
-    }
-
-    public void setPolicyIds(List<String> policyIds) {
-        this.policyIds = policyIds;
-    }
-
-    public String getDedupIntervalMin() {
-        return dedupIntervalMin;
-    }
-
-    public void setDedupIntervalMin(String dedupIntervalMin) {
-        this.dedupIntervalMin = dedupIntervalMin;
-    }
-
-    public Map<String, String> getProperties() {
-        return properties;
-    }
-
-    public void setProperties(Map<String, String> properties) {
-        this.properties = properties;
-    }
-
-    @Override
-    public boolean equals(Object obj) {
-        if (obj instanceof Publishment) {
-            Publishment p = (Publishment) obj;
-            return (Objects.equals(name, p.getName()) &&
-                    Objects.equals(type, p.getType()) &&
-                    Objects.equals(dedupIntervalMin, p.getDedupIntervalMin()) &&
-                    Objects.equals(policyIds, p.getPolicyIds()) &&
-                    properties.equals(p.getProperties()));
-        }
-        return false;
-    }
-
-    @Override
-    public int hashCode() {
-        return new HashCodeBuilder()
-                .append(name)
-                .append(type)
-                .append(dedupIntervalMin)
-                .append(policyIds)
-                .append(properties)
-                .build();
-    }
-
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/coordinator/PublishmentType.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/coordinator/PublishmentType.java b/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/coordinator/PublishmentType.java
deleted file mode 100644
index f34b971..0000000
--- a/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/coordinator/PublishmentType.java
+++ /dev/null
@@ -1,83 +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.coordinator;
-
-import java.util.Objects;
-
-import org.apache.commons.lang3.builder.HashCodeBuilder;
-
-public class PublishmentType {
-
-    private String type;
-    private String className;
-    private String description;
-    private String fields;
-
-    public String getType() {
-        return type;
-    }
-
-    public void setType(String type) {
-        this.type = type;
-    }
-
-    public String getClassName(){
-        return className;
-    }
-    public void setClassName(String className){
-        this.className = className;
-    }
-
-    public String getDescription(){
-        return description;
-    }
-    public void setDescription(String description){
-        this.description = description;
-    }
-
-    public String getFields() {
-        return fields;
-    }
-
-    public void setFields(String fields) {
-        this.fields = fields;
-    }
-
-    @Override
-    public boolean equals(Object obj) {
-        if (obj instanceof PublishmentType) {
-            PublishmentType p = (PublishmentType) obj;
-            return (Objects.equals(className, p.getClassName()) &&
-                    Objects.equals(type, p.type) && 
-                    Objects.equals(description, p.getDescription()) &&
-                    Objects.equals(fields, p.getFields()));
-        }
-        return false;
-    }
-
-    @Override
-    public int hashCode() {
-        return new HashCodeBuilder()
-                .append(className)
-                .append(type)
-                .append(description)
-                .append(fields)
-                .build();
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/coordinator/StreamColumn.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/coordinator/StreamColumn.java b/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/coordinator/StreamColumn.java
deleted file mode 100644
index dc44571..0000000
--- a/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/coordinator/StreamColumn.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.coordinator;
-
-import java.io.Serializable;
-
-import com.fasterxml.jackson.annotation.JsonCreator;
-
-public class StreamColumn implements Serializable{
-    private static final long serialVersionUID = -5457861313624389106L;
-    private String name;
-    private Type type;
-    private Object defaultValue;
-    private boolean required;
-    private String description;
-
-    public String toString(){
-        return String.format("StreamColumn=name[%s], type=[%s], defaultValue=[%s], required=[%s]", name, type, defaultValue, required);
-    }
-
-    public String getName() {
-        return name;
-    }
-
-    public void setName(String name) {
-        this.name = name;
-    }
-
-    public Type getType() {
-        return type;
-    }
-
-    public void setType(Type type) {
-        this.type = type;
-    }
-
-    public Object getDefaultValue() {
-        return defaultValue;
-    }
-
-    public void setDefaultValue(Object defaultValue) {
-        this.defaultValue = defaultValue;
-    }
-
-    public boolean isRequired() {
-        return required;
-    }
-
-    public void setRequired(boolean required) {
-        this.required = required;
-    }
-    
-    public String getDescription() {
-        return description;
-    }
-
-    public void setDescription(String description) {
-        this.description = description;
-    }
-
-
-    public enum Type implements Serializable{
-        STRING("string"), INT("int"), LONG("long"), FLOAT("float"), DOUBLE("double"), BOOL("bool"), OBJECT("object");
-
-        private final String name;
-
-        Type(String name){
-            this.name = name;
-        }
-
-        @Override
-        public String toString() {
-            return name;
-        }
-
-        @JsonCreator
-        public static Type getEnumFromValue(String value) {
-            for (Type testEnum : values()) {
-                if (testEnum.name.equals(value)) {
-                    return testEnum;
-                }
-            }
-            throw new IllegalArgumentException();
-        }
-    }
-
-    public static class Builder {
-        private StreamColumn column;
-
-        public Builder(){
-            column = new StreamColumn();
-        }
-        public Builder name(String name){
-            column.setName(name);
-            return this;
-        }
-        public Builder type(Type type){
-            column.setType(type);
-            return this;
-        }
-        public Builder defaultValue(Object defaultValue){
-            column.setDefaultValue(defaultValue);
-            return this;
-        }
-        public Builder required(boolean required){
-            column.setRequired(required);
-            return this;
-        }
-
-        public StreamColumn build(){
-            return column;
-        }
-    }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/coordinator/StreamDefinition.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/coordinator/StreamDefinition.java b/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/coordinator/StreamDefinition.java
deleted file mode 100644
index cd5773a..0000000
--- a/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/coordinator/StreamDefinition.java
+++ /dev/null
@@ -1,105 +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.coordinator;
-
-import java.io.Serializable;
-import java.util.ArrayList;
-import java.util.List;
-
-/**
- * This is actually a data source schema.
- * 
- * @since Apr 5, 2016
- *
- */
-public class StreamDefinition implements Serializable{
-    private static final long serialVersionUID = 2352202882328931825L;
-    private String streamId;
-    private String dataSource;
-    private String description;
-    private boolean validate;
-    private boolean timeseries;
-
-    private List<StreamColumn> columns = new ArrayList<StreamColumn>();
-
-    public String toString(){
-        return String.format("StreamDefinition[streamId=%s, dataSource=%s, description=%s, validate=%s, timeseries=%s, columns=%s",
-                streamId,
-                dataSource,
-                description,
-                validate,
-                timeseries,
-                columns);
-    }
-
-    public String getStreamId() {
-        return streamId;
-    }
-
-    public void setStreamId(String streamId) {
-        this.streamId = streamId;
-    }
-
-    public String getDescription() {
-        return description;
-    }
-
-    public void setDescription(String description) {
-        this.description = description;
-    }
-
-    public boolean isValidate() {
-        return validate;
-    }
-
-    public void setValidate(boolean validate) {
-        this.validate = validate;
-    }
-
-    public boolean isTimeseries() {
-        return timeseries;
-    }
-
-    public void setTimeseries(boolean timeseries) {
-        this.timeseries = timeseries;
-    }
-
-    public List<StreamColumn> getColumns() {
-        return columns;
-    }
-
-    public void setColumns(List<StreamColumn> columns) {
-        this.columns = columns;
-    }
-
-    public String getDataSource() {
-        return dataSource;
-    }
-
-    public void setDataSource(String dataSource) {
-        this.dataSource = dataSource;
-    }
-
-    public int getColumnIndex(String column){
-        int i=0;
-        for(StreamColumn col:this.getColumns()){
-            if(col.getName().equals(column)) return i;
-            i++;
-        }
-        return -1;
-    }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/coordinator/StreamPartition.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/coordinator/StreamPartition.java b/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/coordinator/StreamPartition.java
deleted file mode 100644
index 7b96024..0000000
--- a/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/coordinator/StreamPartition.java
+++ /dev/null
@@ -1,143 +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.coordinator;
-
-import java.io.Serializable;
-import java.util.*;
-
-import org.apache.commons.collections.CollectionUtils;
-import org.apache.commons.lang3.StringUtils;
-import org.apache.commons.lang3.builder.HashCodeBuilder;
-
-/**
- * StreamPartition defines how a data stream is partitioned and sorted
- * streamId is used for distinguishing different streams which are spawned from the same data source
- * type defines how to partition data among slots within one slotqueue
- * columns are fields based on which stream is grouped
- * sortSpec defines how data is sorted
- */
-public class StreamPartition implements Serializable {
-    private static final long serialVersionUID = -3361648309136926040L;
-
-    private String streamId;
-    private Type type;
-    private List<String> columns = new ArrayList<>();
-    private StreamSortSpec sortSpec;
-
-    public StreamPartition() {
-    }
-
-    public StreamPartition(StreamPartition o) {
-        this.streamId = o.streamId;
-        this.type = o.type;
-        this.columns = new ArrayList<String>(o.columns);
-        this.sortSpec = o.sortSpec == null ? null : new StreamSortSpec(o.sortSpec);
-    }
-
-    @Override
-    public boolean equals(Object other) {
-        if (other == this)
-            return true;
-        if (!(other instanceof StreamPartition)) {
-            return false;
-        }
-        StreamPartition sp = (StreamPartition) other;
-        return Objects.equals(streamId, sp.streamId) && Objects.equals(type, sp.type)
-                && CollectionUtils.isEqualCollection(columns, sp.columns) && Objects.equals(sortSpec, sp.sortSpec);
-    }
-
-    @Override
-    public int hashCode() {
-        return new HashCodeBuilder().append(streamId).append(type).append(columns).append(sortSpec).build();
-    }
-
-    public void setType(Type type) {
-        this.type = type;
-    }
-
-    public Type getType(){
-        return this.type;
-    }
-
-    public enum Type{
-        GLOBAL("GLOBAL",0), GROUPBY("GROUPBY",1), SHUFFLE("SHUFFLE",2);
-        private final String name;
-        private final int index;
-        Type(String name, int index){
-            this.name = name;
-            this.index = index;
-        }
-        @Override
-        public String toString() {
-            return this.name;
-        }
-        public static Type locate(String type){
-            Type _type = _NAME_TYPE.get(type.toUpperCase());
-            if(_type == null)
-                throw new IllegalStateException("Illegal type name: "+type);
-            return _type;
-        }
-
-        public static Type locate(int index){
-            Type _type = _INDEX_TYPE.get(index);
-            if(_type == null)
-                throw new IllegalStateException("Illegal type index: "+index);
-            return _type;
-        }
-
-        private static final Map<String,Type> _NAME_TYPE = new HashMap<>();
-        private static final Map<Integer,Type> _INDEX_TYPE = new TreeMap<>();
-        static {
-            _NAME_TYPE.put(GLOBAL.name,GLOBAL);
-            _NAME_TYPE.put(GROUPBY.name,GROUPBY);
-            _NAME_TYPE.put(SHUFFLE.name,SHUFFLE);
-
-            _INDEX_TYPE.put(GLOBAL.index,GLOBAL);
-            _INDEX_TYPE.put(GROUPBY.index,GLOBAL);
-            _INDEX_TYPE.put(SHUFFLE.index,GLOBAL);
-        }
-    }
-
-    public List<String> getColumns() {
-        return columns;
-    }
-
-    public void setColumns(List<String> columns) {
-        this.columns = columns;
-    }
-
-    public String getStreamId() {
-        return streamId;
-    }
-
-    public void setStreamId(String streamId) {
-        this.streamId = streamId;
-    }
-
-    public StreamSortSpec getSortSpec() {
-        return sortSpec;
-    }
-
-    public void setSortSpec(StreamSortSpec sortSpec) {
-        this.sortSpec = sortSpec;
-    }
-
-    @Override
-    public String toString() {
-        return String.format("StreamPartition[streamId=%s,type=%s,columns=[%s],sortSpec=[%s]]",this.getStreamId(),this.getType(), StringUtils.join(this.getColumns(),","), sortSpec);
-    }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/coordinator/StreamSortSpec.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/coordinator/StreamSortSpec.java b/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/coordinator/StreamSortSpec.java
deleted file mode 100644
index ee20f81..0000000
--- a/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/coordinator/StreamSortSpec.java
+++ /dev/null
@@ -1,98 +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.coordinator;
-
-
-import java.io.Serializable;
-
-import org.apache.commons.lang.builder.HashCodeBuilder;
-import org.apache.eagle.alert.utils.TimePeriodUtils;
-import org.joda.time.Period;
-
-/**
- * streamId is the key
- */
-public class StreamSortSpec implements Serializable{
-    private static final long serialVersionUID = 3626506441441584937L;
-    private String windowPeriod="";
-    private int windowMargin = 30 * 1000; // 30 seconds by default
-
-    public StreamSortSpec() {}
-
-    public StreamSortSpec(StreamSortSpec spec) {
-        this.windowPeriod = spec.windowPeriod;
-        this.windowMargin = spec.windowMargin;
-    }
-
-    public String getWindowPeriod() {
-        return windowPeriod;
-    }
-
-    public int getWindowPeriodMillis() {
-        if(windowPeriod!=null) {
-            return TimePeriodUtils.getMillisecondsOfPeriod(Period.parse(windowPeriod));
-        } else return 0;
-    }
-
-    public void setWindowPeriod(String windowPeriod) {
-        this.windowPeriod = windowPeriod;
-    }
-    public void setWindowPeriodMillis(int windowPeriodMillis) {
-        this.windowPeriod = Period.millis(windowPeriodMillis).toString();
-    }
-
-    public void setWindowPeriod2(Period period) {
-        this.windowPeriod = period.toString();
-    }
-
-
-    public int getWindowMargin() {
-        return windowMargin;
-    }
-
-    public void setWindowMargin(int windowMargin) {
-        this.windowMargin = windowMargin;
-    }
-
-    @Override
-    public int hashCode(){
-        return new HashCodeBuilder().
-                append(windowPeriod).
-                append(windowMargin).toHashCode();
-    }
-
-    @Override
-    public boolean equals(Object that){
-        if(this == that)
-            return true;
-        if(!(that instanceof StreamSortSpec)){
-            return false;
-        }
-
-        StreamSortSpec another = (StreamSortSpec)that;
-        return 
-                another.windowPeriod.equals(this.windowPeriod) &&
-                another.windowMargin == this.windowMargin;
-    }
-
-    @Override
-    public String toString(){
-        return String.format("StreamSortSpec[windowPeriod=%s,windowMargin=%d]",
-                this.getWindowPeriod(),
-                this.getWindowMargin());
-    }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/coordinator/StreamingCluster.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/coordinator/StreamingCluster.java b/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/coordinator/StreamingCluster.java
deleted file mode 100644
index 6cafb16..0000000
--- a/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/coordinator/StreamingCluster.java
+++ /dev/null
@@ -1,89 +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.coordinator;
-
-import java.util.Map;
-
-import com.fasterxml.jackson.annotation.JsonProperty;
-
-/**
- * @since Apr 5, 2016
- *
- */
-public class StreamingCluster {
-    public static enum StreamingType {
-        STORM
-    }
-
-    @JsonProperty
-    private String name;
-    @JsonProperty
-    private String zone;
-    @JsonProperty
-    private StreamingType type;
-    @JsonProperty
-    private String description;
-    /**
-     * key - nimbus for storm
-     */
-    @JsonProperty
-    private Map<String, String> deployments;
-
-    public String getName() {
-        return name;
-    }
-
-    public void setName(String name) {
-        this.name = name;
-    }
-
-    public String getZone() {
-        return zone;
-    }
-
-    public void setZone(String zone) {
-        this.zone = zone;
-    }
-
-    public StreamingType getType() {
-        return type;
-    }
-
-    public void setType(StreamingType type) {
-        this.type = type;
-    }
-
-    public String getDescription() {
-        return description;
-    }
-
-    public void setDescription(String description) {
-        this.description = description;
-    }
-
-    public Map<String, String> getDeployments() {
-        return deployments;
-    }
-
-    public void setDeployments(Map<String, String> deployments) {
-        this.deployments = deployments;
-    }
-
-    public static final String NIMBUS_HOST = "nimbusHost";
-    public static final String NIMBUS_THRIFT_PORT = "nimbusThriftPort";
-
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/model/AlertStreamEvent.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/model/AlertStreamEvent.java b/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/model/AlertStreamEvent.java
deleted file mode 100644
index f36d3cb..0000000
--- a/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/model/AlertStreamEvent.java
+++ /dev/null
@@ -1,89 +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.model;
-
-import java.util.ArrayList;
-import java.util.List;
-
-import org.apache.commons.lang3.StringUtils;
-import org.apache.eagle.alert.engine.coordinator.PolicyDefinition;
-import org.apache.eagle.alert.engine.coordinator.StreamDefinition;
-import org.apache.eagle.alert.utils.DateTimeUtil;
-
-/**
- * streamId stands for alert type instead of source event streamId
- */
-public class AlertStreamEvent extends StreamEvent {
-    private static final long serialVersionUID = 2392131134670106397L;
-
-    // TODO: Keep policy name only instead of policy entity
-    private PolicyDefinition policy;
-    private StreamDefinition schema;
-    private String createdBy;
-    private long createdTime;
-
-    public PolicyDefinition getPolicy() {
-        return policy;
-    }
-
-    public void setPolicy(PolicyDefinition policy) {
-        this.policy = policy;
-    }
-
-
-    public String getPolicyId() {
-        return policy.getName();
-    }
-
-    @Override
-    public String toString() {
-        List<String> dataStrings = new ArrayList<>(this.getData().length);
-        for(Object obj: this.getData()){
-            if(obj!=null) {
-                dataStrings.add(obj.toString());
-            }else{
-                dataStrings.add(null);
-            }
-        }
-        return String.format("AlertStreamEvent[stream=%S,timestamp=%s,data=[%s], policy=%s, createdBy=%s]",
-                this.getStreamId(), DateTimeUtil.millisecondsToHumanDateWithMilliseconds(this.getTimestamp()), StringUtils.join(dataStrings,","),this.getPolicy().getName(),this.getCreatedBy());
-    }
-
-    public String getCreatedBy() {
-        return createdBy;
-    }
-
-    public void setCreatedBy(String createdBy) {
-        this.createdBy = createdBy;
-    }
-
-    public StreamDefinition getSchema() {
-        return schema;
-    }
-
-    public void setSchema(StreamDefinition schema) {
-        this.schema = schema;
-    }
-
-    public long getCreatedTime() {
-        return createdTime;
-    }
-
-    public void setCreatedTime(long createdTime) {
-        this.createdTime = createdTime;
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/model/PartitionedEvent.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/model/PartitionedEvent.java b/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/model/PartitionedEvent.java
deleted file mode 100644
index cfed3e2..0000000
--- a/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/model/PartitionedEvent.java
+++ /dev/null
@@ -1,149 +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.model;
-
-import java.io.Serializable;
-import java.util.Objects;
-
-import org.apache.commons.lang3.builder.HashCodeBuilder;
-import org.apache.eagle.alert.engine.coordinator.StreamPartition;
-
-import backtype.storm.tuple.Tuple;
-
-/**
- * This is a critical data structure across spout, router bolt and alert bolt
- * partition[StreamPartition] defines how one incoming data stream is partitioned, sorted
- * partitionKey[long] is java hash value of groupby fields. The groupby fields are defined in StreamPartition
- * event[StreamEvent] is actual data
- */
-public class PartitionedEvent implements Serializable{
-    private static final long serialVersionUID = -3840016190614238593L;
-    private StreamPartition partition;
-    private long partitionKey;
-    private StreamEvent event;
-
-    /**
-     * Used for bolt-internal but not inter-bolts,
-     * will not pass across bolts
-     */
-    private transient Tuple anchor;
-
-    public PartitionedEvent(){
-        this.event = null;
-        this.partition = null;
-        this.partitionKey = 0L;
-    }
-
-    public PartitionedEvent(StreamEvent event, StreamPartition partition, int partitionKey) {
-        this.event = event;
-        this.partition = partition;
-        this.partitionKey = partitionKey;
-    }
-
-    @Override
-    public boolean equals(Object obj) {
-        if(obj == this) return true;
-        if(obj == null) return false;
-        if(obj instanceof PartitionedEvent){
-            PartitionedEvent another = (PartitionedEvent) obj;
-            return !(this.partitionKey != another.getPartitionKey()
-                    || !Objects.equals(this.event, another.getEvent())
-                    || !Objects.equals(this.partition, another.getPartition())
-                    || !Objects.equals(this.anchor, another.anchor));
-        } else {
-            return false;
-        }
-    }
-
-    @Override
-    public int hashCode() {
-        return new HashCodeBuilder()
-                .append(partitionKey)
-                .append(event)
-                .append(partition)
-                .build();
-    }
-
-    public StreamEvent getEvent() {
-        return event;
-    }
-
-    public void setEvent(StreamEvent event) {
-        this.event = event;
-    }
-
-    public StreamPartition getPartition() {
-        return partition;
-    }
-
-    public void setPartition(StreamPartition partition) {
-        this.partition = partition;
-    }
-
-    public void setPartitionKey(long partitionKey){
-        this.partitionKey = partitionKey;
-    }
-
-    public long getPartitionKey(){
-        return this.partitionKey;
-    }
-
-    public String toString(){
-       return String.format("PartitionedEvent[partition=%s,event=%s,key=%s", partition, event,partitionKey);
-    }
-
-    public long getTimestamp() {
-        return (event != null) ? event.getTimestamp() : 0L;
-    }
-
-    public String getStreamId(){
-        return (event != null) ? event.getStreamId() : null;
-    }
-
-    public Object[] getData(){
-        return event!=null ? event.getData() : null;
-    }
-
-    public boolean isSortRequired(){
-        return isPartitionRequired() && this.getPartition().getSortSpec()!=null;
-    }
-
-    public boolean isPartitionRequired(){
-        return this.getPartition()!=null;
-    }
-
-    public PartitionedEvent copy() {
-        PartitionedEvent copied = new PartitionedEvent();
-        copied.setEvent(this.getEvent());
-        copied.setPartition(this.partition);
-        copied.setPartitionKey(this.partitionKey);
-        return copied;
-    }
-
-    public Tuple getAnchor() {
-        return anchor;
-    }
-
-    public void setAnchor(Tuple anchor) {
-        this.anchor = anchor;
-    }
-
-    public PartitionedEvent withAnchor(Tuple tuple){
-        this.setAnchor(tuple);
-        return this;
-    }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/model/StreamEvent.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/model/StreamEvent.java b/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/model/StreamEvent.java
deleted file mode 100644
index 3e4e1df..0000000
--- a/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/model/StreamEvent.java
+++ /dev/null
@@ -1,144 +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.model;
-
-import java.io.Serializable;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
-import java.util.Objects;
-
-import org.apache.commons.lang3.StringUtils;
-import org.apache.commons.lang3.builder.HashCodeBuilder;
-import org.apache.eagle.alert.engine.coordinator.StreamDefinition;
-import org.apache.eagle.alert.utils.DateTimeUtil;
-
-/**
- * @since Apr 5, 2016
- *
- */
-public class StreamEvent implements Serializable {
-    private static final long serialVersionUID = 2765116509856609763L;
-
-    private String streamId;
-    private Object[] data;
-    private long timestamp;
-
-    public StreamEvent(){}
-
-    public StreamEvent(String streamId,long timestamp,Object[] data){
-        this.setStreamId(streamId);
-        this.setTimestamp(timestamp);
-        this.setData(data);
-    }
-
-    public String getStreamId() {
-        return streamId;
-    }
-
-    public void setStreamId(String streamId) {
-        this.streamId = streamId;
-    }
-
-    public Object[] getData() {
-        return data;
-    }
-
-    public void setData(Object[] data) {
-        this.data = data;
-    }
-
-    public long getTimestamp() {
-        return timestamp;
-    }
-
-    public void setTimestamp(long timestamp) {
-        this.timestamp = timestamp;
-    }
-
-    @Override
-    public int hashCode() {
-        return new HashCodeBuilder().append(streamId).append(timestamp).append(data).build();
-    }
-
-    @Override
-    public boolean equals(Object obj) {
-        if(obj == this) return true;
-        if(obj instanceof StreamEvent){
-            StreamEvent another = (StreamEvent) obj;
-            return Objects.equals(this.streamId,another.streamId) && this.timestamp == another.timestamp && Arrays.deepEquals(this.data,another.data);
-        }
-        return false;
-    }
-
-    @Override
-    public String toString() {
-        List<String> dataStrings = new ArrayList<>();
-        if(this.getData() != null) {
-            for (Object obj : this.getData()) {
-                if (obj != null) {
-                    dataStrings.add(obj.toString());
-                } else {
-                    dataStrings.add(null);
-                }
-            }
-        }
-        return String.format("StreamEvent[stream=%S,timestamp=%s,data=[%s]]",this.getStreamId(), DateTimeUtil.millisecondsToHumanDateWithMilliseconds(this.getTimestamp()), StringUtils.join(dataStrings,","));
-    }
-
-    public static StreamEventBuilder Builder(){
-        return new StreamEventBuilder();
-    }
-
-    /**
-     * @return cloned new event object
-     */
-    public StreamEvent copy(){
-        StreamEvent newEvent = new StreamEvent();
-        newEvent.setTimestamp(this.getTimestamp());
-        newEvent.setData(this.getData());
-        newEvent.setStreamId(this.getStreamId());
-        return newEvent;
-    }
-
-    public void copyFrom(StreamEvent event){
-        this.setTimestamp(event.getTimestamp());
-        this.setData(event.getData());
-        this.setStreamId(event.getStreamId());
-    }
-
-    /**
-     * @param column
-     * @param streamDefinition
-     * @return
-     */
-    public Object[] getData(StreamDefinition streamDefinition,List<String> column) {
-        ArrayList<Object> result = new ArrayList<>(column.size());
-        for (String colName : column) {
-            result.add(this.getData()[streamDefinition.getColumnIndex(colName)]);
-        }
-        return result.toArray();
-    }
-
-    public Object[] getData(StreamDefinition streamDefinition,String ... column) {
-        ArrayList<Object> result = new ArrayList<>(column.length);
-        for (String colName : column) {
-            result.add(this.getData()[streamDefinition.getColumnIndex(colName)]);
-        }
-        return result.toArray();
-    }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/model/StreamEventBuilder.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/model/StreamEventBuilder.java b/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/model/StreamEventBuilder.java
deleted file mode 100644
index 136fd8b..0000000
--- a/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/engine/model/StreamEventBuilder.java
+++ /dev/null
@@ -1,88 +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.model;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-
-import org.apache.eagle.alert.engine.coordinator.StreamColumn;
-import org.apache.eagle.alert.engine.coordinator.StreamDefinition;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class StreamEventBuilder{
-    private final static Logger LOG = LoggerFactory.getLogger(StreamEventBuilder.class);
-
-    private StreamEvent instance;
-    private StreamDefinition streamDefinition;
-    public StreamEventBuilder(){
-        instance = new StreamEvent();
-    }
-
-    public StreamEventBuilder schema(StreamDefinition streamDefinition){
-        this.streamDefinition = streamDefinition;
-        if(instance.getStreamId() == null) instance.setStreamId(streamDefinition.getStreamId());
-        return this;
-    }
-
-    public StreamEventBuilder streamId(String streamId){
-        instance.setStreamId(streamId);
-        return this;
-    }
-
-    public StreamEventBuilder attributes(Map<String,Object> data, StreamDefinition streamDefinition){
-        this.schema(streamDefinition);
-        List<StreamColumn> columnList = streamDefinition.getColumns();
-        if(columnList!=null && columnList.size() > 0){
-            List<Object> values = new ArrayList<>(columnList.size());
-            for (StreamColumn column : columnList) {
-                values.add(data.getOrDefault(column.getName(),column.getDefaultValue()));
-            }
-            instance.setData(values.toArray());
-        } else if(LOG.isDebugEnabled()){
-            LOG.warn("All data [{}] are ignored as no columns defined in schema {}",data,streamDefinition);
-        }
-        return this;
-    }
-
-    public StreamEventBuilder attributes(Map<String,Object> data){
-        return attributes(data,this.streamDefinition);
-    }
-
-    public StreamEventBuilder attributes(Object ... data){
-        instance.setData(data);
-        return this;
-    }
-
-    public StreamEventBuilder timestamep(long timestamp){
-        instance.setTimestamp(timestamp);
-        return this;
-    }
-
-    public StreamEvent build(){
-        if(instance.getStreamId() == null){
-            throw new IllegalArgumentException("streamId is null of event: " + instance);
-        }
-        return instance;
-    }
-
-    public StreamEventBuilder copyFrom(StreamEvent event) {
-        this.instance.copyFrom(event);
-        return this;
-    }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/metric/IMetricSystem.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/metric/IMetricSystem.java b/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/metric/IMetricSystem.java
deleted file mode 100644
index 06a99f4..0000000
--- a/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/metric/IMetricSystem.java
+++ /dev/null
@@ -1,64 +0,0 @@
-package org.apache.eagle.alert.metric;
-
-import java.util.Map;
-
-import org.apache.eagle.alert.metric.sink.MetricSink;
-import org.apache.eagle.alert.metric.source.MetricSource;
-
-import com.codahale.metrics.MetricRegistry;
-import com.typesafe.config.Config;
-
-/**
- * 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.
- */
-public interface IMetricSystem {
-
-    /**
-     * Initialize
-     */
-    void start();
-
-    /**
-     * Schedule reporter
-     */
-    void schedule();
-
-    /**
-     * Close and stop all resources and services
-     */
-    void stop();
-
-    /**
-     * Manual report metric
-     */
-    void report();
-
-    /**
-     *
-     * @param sink metric sink
-     */
-    void register(MetricSink sink,Config config);
-
-    /**
-     *
-     * @param source metric source
-     */
-    void register(MetricSource source);
-
-    void tags(Map<String,Object> metricTags);
-
-    MetricRegistry registry();
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/metric/MetricSystem.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/metric/MetricSystem.java b/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/metric/MetricSystem.java
deleted file mode 100644
index b91c606..0000000
--- a/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/metric/MetricSystem.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.metric;
-
-import java.util.HashMap;
-import java.util.Map;
-import java.util.concurrent.TimeUnit;
-
-import org.apache.eagle.alert.metric.sink.MetricSink;
-import org.apache.eagle.alert.metric.sink.MetricSinkRepository;
-import org.apache.eagle.alert.metric.source.MetricSource;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.codahale.metrics.MetricRegistry;
-import com.typesafe.config.Config;
-import com.typesafe.config.ConfigFactory;
-
-public class MetricSystem implements IMetricSystem {
-    private final Config config;
-    private Map<MetricSink,Config> sinks = new HashMap<>();
-//    private Map<String,MetricSource> sources = new HashMap<>();
-    private MetricRegistry registry = new MetricRegistry();
-    private boolean running;
-    private boolean initialized;
-    private final static Logger LOG = LoggerFactory.getLogger(MetricSystem.class);
-    private final Map<String, Object> metricTags = new HashMap<>();
-
-    public MetricSystem(Config config){
-        this.config = config;
-    }
-
-    public static MetricSystem load(Config config){
-        MetricSystem instance = new MetricSystem(config);
-        instance.loadFromConfig();
-        return instance;
-    }
-
-    @Override
-    public void tags(Map<String,Object> metricTags){
-        this.metricTags.putAll(metricTags);
-    }
-
-    @Override
-    public void start() {
-        if(initialized)
-            throw new IllegalStateException("Attempting to initialize a MetricsSystem that is already intialized");
-        sinks.forEach((sink,conf) -> sink.prepare(conf.withValue("tags",ConfigFactory.parseMap(metricTags).root()),registry));
-        initialized = true;
-    }
-
-    @Override
-    public void schedule() {
-        if(running){
-           throw  new IllegalStateException("Attempting to start a MetricsSystem that is already running");
-        }
-
-        sinks.keySet().forEach((sink)->sink.start(5, TimeUnit.SECONDS));
-        running = true;
-    }
-
-    public void loadFromConfig(){
-        loadSinksFromConfig();
-    }
-
-    private void loadSinksFromConfig(){
-        Config sinkCls = config.hasPath("metric.sink") ? config.getConfig("metric.sink") : null;
-        if(sinkCls == null){
-            // do nothing
-        }else{
-            for(String sinkType:sinkCls.root().unwrapped().keySet()){
-                register(MetricSinkRepository.createSink(sinkType),config.getConfig("metric.sink."+sinkType));
-            }
-        }
-    }
-
-    @Override
-    public void stop() {
-        sinks.keySet().forEach(MetricSink::stop);
-    }
-
-    @Override
-    public void report() {
-        sinks.keySet().forEach(MetricSink::report);
-    }
-
-    @Override
-    public void register(MetricSink sink,Config config) {
-        LOG.debug("Register {}",sink);
-        sinks.put(sink,config);
-    }
-
-    @Override
-    public void register(MetricSource source) {
-        registry().registerAll(source.registry());
-    }
-
-    @Override
-    public MetricRegistry registry() {
-        return registry;
-    }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/metric/entity/MetricEvent.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/metric/entity/MetricEvent.java b/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/metric/entity/MetricEvent.java
deleted file mode 100644
index b5e6c63..0000000
--- a/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/metric/entity/MetricEvent.java
+++ /dev/null
@@ -1,111 +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.metric.entity;
-
-import java.util.Map;
-import java.util.TreeMap;
-
-import org.apache.eagle.alert.utils.DateTimeUtil;
-
-import com.codahale.metrics.Counter;
-import com.codahale.metrics.Gauge;
-import com.codahale.metrics.Histogram;
-import com.codahale.metrics.Meter;
-import com.codahale.metrics.Snapshot;
-import com.codahale.metrics.Timer;
-
-public class MetricEvent extends TreeMap<String,Object>{
-
-    private static final long serialVersionUID = 6862373651636342744L;
-
-    public static Builder of(String name){
-        return new Builder(name);
-    }
-
-    /**
-     * TODO: Refactor according to ConsoleReporter
-     */
-    public static class Builder{
-        private final String name;
-        private MetricEvent instance;
-        public Builder(String name){
-            this.instance = new MetricEvent();
-            this.name = name;
-        }
-
-        public Builder from(Counter value) {
-//            this.instance.put("type","counter");
-            this.instance.put("count",value.getCount());
-            return this;
-        }
-
-        public MetricEvent build(){
-            this.instance.put("name",name);
-            if(!this.instance.containsKey("timestamp")){
-                this.instance.put("timestamp", DateTimeUtil.getCurrentTimestamp());
-            }
-            return this.instance;
-        }
-
-        @SuppressWarnings({ "rawtypes", "unchecked" })
-        public Builder from(Gauge gauge) {
-            Object value = gauge.getValue();
-            if( value instanceof Map){
-                Map<? extends String, ?> map = (Map<? extends String, ?>) value;
-                this.instance.putAll(map);
-            } else {
-                this.instance.put("value", value);
-            }
-            return this;
-        }
-
-        public Builder from(Histogram value) {
-            this.instance.put("count",value.getCount());
-            Snapshot snapshot = value.getSnapshot();
-            this.instance.put("min", snapshot.getMin());
-            this.instance.put("max", snapshot.getMax());
-            this.instance.put("mean", snapshot.getMean());
-            this.instance.put("stddev", snapshot.getStdDev());
-            this.instance.put("median", snapshot.getMedian());
-            this.instance.put("75thPercentile", snapshot.get75thPercentile());
-            this.instance.put("95thPercentile", snapshot.get95thPercentile());
-            this.instance.put("98thPercentile", snapshot.get98thPercentile());
-            this.instance.put("99thPercentile", snapshot.get99thPercentile());
-            this.instance.put("999thPercentile", snapshot.get999thPercentile());
-            return this;
-        }
-
-        public Builder from(Meter value) {
-            this.instance.put("value",value.getCount());
-            this.instance.put("15MinRate",value.getFifteenMinuteRate());
-            this.instance.put("5MinRate",value.getFiveMinuteRate());
-            this.instance.put("mean",value.getMeanRate());
-            this.instance.put("1MinRate",value.getOneMinuteRate());
-            return this;
-        }
-
-        public Builder from(Timer value) {
-//            this.instance.put("type","timer");
-            this.instance.put("value",value.getCount());
-            this.instance.put("15MinRate",value.getFifteenMinuteRate());
-            this.instance.put("5MinRate",value.getFiveMinuteRate());
-            this.instance.put("mean",value.getMeanRate());
-            this.instance.put("1MinRate",value.getOneMinuteRate());
-            return this;
-        }
-    }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/metric/reporter/KafkaReporter.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/metric/reporter/KafkaReporter.java b/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/metric/reporter/KafkaReporter.java
deleted file mode 100644
index 42bf5d5..0000000
--- a/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/metric/reporter/KafkaReporter.java
+++ /dev/null
@@ -1,213 +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.metric.reporter;
-
-import java.util.Map;
-import java.util.Properties;
-import java.util.SortedMap;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-
-import org.apache.eagle.alert.metric.entity.MetricEvent;
-import org.apache.eagle.alert.utils.ByteUtils;
-import org.apache.kafka.clients.producer.KafkaProducer;
-import org.apache.kafka.clients.producer.Producer;
-import org.apache.kafka.clients.producer.ProducerRecord;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.codahale.metrics.ConsoleReporter;
-import com.codahale.metrics.Counter;
-import com.codahale.metrics.Gauge;
-import com.codahale.metrics.Histogram;
-import com.codahale.metrics.Meter;
-import com.codahale.metrics.MetricFilter;
-import com.codahale.metrics.MetricRegistry;
-import com.codahale.metrics.ScheduledReporter;
-import com.codahale.metrics.Timer;
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.google.common.base.Preconditions;
-import com.typesafe.config.Config;
-
-public class KafkaReporter extends ScheduledReporter {
-	private final static Logger LOG = LoggerFactory.getLogger(KafkaReporter.class);
-	private final String topic;
-	private final Properties properties;
-	private final Producer<byte[], String> producer;
-	private final Map<String, Object> additionalFields;
-
-	protected KafkaReporter(MetricRegistry registry, MetricFilter filter, TimeUnit rateUnit, TimeUnit durationUnit, String topic, Properties config, Map<String, Object> additionalFields) {
-		super(registry, "kafka-reporter", filter, rateUnit, durationUnit);
-		this.topic = topic;
-		this.properties = new Properties();
-		Preconditions.checkNotNull(topic,"topic should not be null");
-//		properties.put("bootstrap.servers", brokerList);
-//		properties.put("metadata.broker.list", brokerList);
-		properties.put("key.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
-		properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
-		properties.put("request.required.acks", "1");
-		properties.put("key.deserializer","org.apache.kafka.common.serialization.ByteArraySerializer");
-		properties.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
-		if(config != null) {
-			LOG.info(config.toString());
-			properties.putAll(config);
-		}
-		this.additionalFields = additionalFields;
-		this.producer = new KafkaProducer<>(properties);
-		LOG.info("Initialized kafka-reporter");
-	}
-
-	@SuppressWarnings("rawtypes")
-    @Override
-	public void report(SortedMap<String, Gauge> gauges, SortedMap<String, Counter> counters, SortedMap<String, Histogram> histograms, SortedMap<String, Meter> meters, SortedMap<String, Timer> timers) {
-		for(SortedMap.Entry<String, Gauge> entry:gauges.entrySet()){
-			onMetricEvent(MetricEvent.of(entry.getKey()).from(entry.getValue()).build());
-		}
-		for(SortedMap.Entry<String, Counter> entry:counters.entrySet()){
-			onMetricEvent(MetricEvent.of(entry.getKey()).from(entry.getValue()).build());
-		}
-		for(SortedMap.Entry<String, Histogram> entry:histograms.entrySet()){
-			onMetricEvent(MetricEvent.of(entry.getKey()).from(entry.getValue()).build());
-		}
-		for(SortedMap.Entry<String, Meter> entry:meters.entrySet()){
-			onMetricEvent(MetricEvent.of(entry.getKey()).from(entry.getValue()).build());
-		}
-		for(SortedMap.Entry<String, Timer> entry:timers.entrySet()){
-			onMetricEvent(MetricEvent.of(entry.getKey()).from(entry.getValue()).build());
-		}
-	}
-
-	private final static ObjectMapper OBJECT_MAPPER = new ObjectMapper();
-
-	private void onMetricEvent(MetricEvent event){
-		try {
-			if(additionalFields!=null){
-				event.putAll(additionalFields);
-			}
-			// TODO: Support configurable partition key
-			byte[] key = ByteUtils.intToBytes(event.hashCode());
-			ProducerRecord<byte[],String> record  = new ProducerRecord<>(topic, key, OBJECT_MAPPER.writeValueAsString(event));
-			// TODO: Support configuration timeout
-			this.producer.send(record).get(5,TimeUnit.SECONDS);
-		} catch (JsonProcessingException e) {
-			LOG.error("Failed to serialize {} as json",event,e);
-		} catch (InterruptedException | ExecutionException | TimeoutException e) {
-			LOG.error("Failed to produce message to topic {}",topic,e);
-		}
-	}
-
-	@Override
-	public void stop() {
-		this.producer.close();
-		super.stop();
-	}
-
-	@Override
-	public void close() {
-		this.producer.close();
-		super.close();
-	}
-
-	public static Builder forRegistry(MetricRegistry registry){
-		return new Builder(registry);
-	}
-
-	public static class Builder{
-		private final MetricRegistry registry;
-		private TimeUnit rateUnit;
-		private TimeUnit durationUnit;
-		private MetricFilter filter;
-		private String topic;
-		private Properties properties;
-		private Map<String, Object> additionalFields;
-
-		private Builder(MetricRegistry registry) {
-			this.registry = registry;
-			this.rateUnit = TimeUnit.SECONDS;
-			this.durationUnit = TimeUnit.MILLISECONDS;
-			this.filter = MetricFilter.ALL;
-		}
-
-		/**
-		 * Convert rates to the given time unit.
-		 *
-		 * @param rateUnit a unit of time
-		 * @return {@code this}
-		 */
-		public Builder convertRatesTo(TimeUnit rateUnit) {
-			this.rateUnit = rateUnit;
-			return this;
-		}
-
-		/**
-		 * Convert durations to the given time unit.
-		 *
-		 * @param durationUnit a unit of time
-		 * @return {@code this}
-		 */
-		public Builder convertDurationsTo(TimeUnit durationUnit) {
-			this.durationUnit = durationUnit;
-			return this;
-		}
-
-		/**
-		 * Only report metrics which match the given filter.
-		 *
-		 * @param filter a {@link MetricFilter}
-		 * @return {@code this}
-		 */
-		public Builder filter(MetricFilter filter) {
-			this.filter = filter;
-			return this;
-		}
-
-		public Builder topic(String topic){
-			this.topic = topic;
-			return this;
-		}
-
-		public Builder config(Properties properties){
-			this.properties = properties;
-			return this;
-		}
-
-		/**
-		 * Builds a {@link ConsoleReporter} with the given properties.
-		 *
-		 * @return a {@link ConsoleReporter}
-		 */
-		public KafkaReporter build() {
-			if(topic == null && properties!=null) topic = properties.getProperty("topic");
-			return new KafkaReporter(registry,filter,rateUnit,durationUnit,topic,properties,additionalFields);
-		}
-
-		@SuppressWarnings("serial")
-        public Builder config(Config config) {
-			this.config(new Properties(){{
-				putAll(config.root().unwrapped());
-			}});
-			return this;
-		}
-
-		public Builder addFields(Map<String, Object> tags) {
-			this.additionalFields = tags;
-			return this;
-		}
-	}
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/metric/sink/ConsoleSink.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/metric/sink/ConsoleSink.java b/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/metric/sink/ConsoleSink.java
deleted file mode 100644
index fd6cc41..0000000
--- a/eagle-core/eagle-alert/alert/alert-common/src/main/java/org/apache/eagle/alert/metric/sink/ConsoleSink.java
+++ /dev/null
@@ -1,47 +0,0 @@
-package org.apache.eagle.alert.metric.sink;
-
-import java.util.concurrent.TimeUnit;
-
-import com.codahale.metrics.ConsoleReporter;
-import com.codahale.metrics.MetricRegistry;
-import com.typesafe.config.Config;
-
-/**
- * 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.
- */
-public class ConsoleSink implements MetricSink {
-    private ConsoleReporter reporter;
-    @Override
-    public void prepare(Config config, MetricRegistry registry) {
-        reporter = ConsoleReporter.forRegistry(registry).build();
-    }
-
-    @Override
-    public void start(long period,TimeUnit unit) {
-        reporter.start(period, unit);
-    }
-
-    @Override
-    public void stop() {
-        reporter.stop();
-        reporter.close();
-    }
-
-    @Override
-    public void report() {
-        reporter.report();
-    }
-}
\ No newline at end of file


Mime
View raw message