eagle-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From h..@apache.org
Subject [12/46] incubator-eagle git commit: [EAGLE-325] Initialize next-gen alert engine code on branch-0.5
Date Thu, 02 Jun 2016 07:07:51 GMT
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/sorter/impl/PartitionedEventTimeOrderingComparator.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/sorter/impl/PartitionedEventTimeOrderingComparator.java b/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/sorter/impl/PartitionedEventTimeOrderingComparator.java
deleted file mode 100644
index a9f9f39..0000000
--- a/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/sorter/impl/PartitionedEventTimeOrderingComparator.java
+++ /dev/null
@@ -1,50 +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.sorter.impl;
-
-import java.util.Comparator;
-import java.util.Objects;
-
-import org.apache.eagle.alert.engine.model.PartitionedEvent;
-
-/**
- * TODO: Stable sorting algorithm for better performance to avoid event resorting with same timestamp?
- */
-public class PartitionedEventTimeOrderingComparator implements Comparator<PartitionedEvent> {
-    public static final PartitionedEventTimeOrderingComparator INSTANCE = new PartitionedEventTimeOrderingComparator();
-
-    @Override
-    public int compare(PartitionedEvent o1, PartitionedEvent o2) {
-        if(Objects.equals(o1,o2)){
-            return 0;
-        }else {
-            if(o1 == null && o2 == null){
-                return 0;
-            }else if(o1 != null && o2 == null){
-                return 1;
-            }else if(o1 == null){
-                return -1;
-            }
-            // Unstable Sorting Algorithm
-            if(o1.getTimestamp() <= o2.getTimestamp()){
-                return -1;
-            } else {
-                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-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/sorter/impl/StreamSortWindowHandlerImpl.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/sorter/impl/StreamSortWindowHandlerImpl.java b/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/sorter/impl/StreamSortWindowHandlerImpl.java
deleted file mode 100644
index 7be69e1..0000000
--- a/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/sorter/impl/StreamSortWindowHandlerImpl.java
+++ /dev/null
@@ -1,113 +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.sorter.impl;
-
-import java.io.IOException;
-
-import org.apache.eagle.alert.engine.PartitionedEventCollector;
-import org.apache.eagle.alert.engine.coordinator.StreamSortSpec;
-import org.apache.eagle.alert.engine.model.PartitionedEvent;
-import org.apache.eagle.alert.engine.router.StreamSortHandler;
-import org.apache.eagle.alert.engine.sorter.StreamTimeClock;
-import org.apache.eagle.alert.engine.sorter.StreamWindow;
-import org.apache.eagle.alert.engine.sorter.StreamWindowManager;
-import org.apache.eagle.alert.utils.DateTimeUtil;
-import org.joda.time.Period;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class StreamSortWindowHandlerImpl implements StreamSortHandler {
-    private final static Logger LOG = LoggerFactory.getLogger(StreamSortWindowHandlerImpl.class);
-    private StreamWindowManager windowManager;
-    private StreamSortSpec streamSortSpecSpec;
-    private PartitionedEventCollector outputCollector;
-    private String streamId;
-
-    public void prepare(String streamId, StreamSortSpec streamSortSpecSpec, PartitionedEventCollector outputCollector) {
-        this.windowManager = new StreamWindowManagerImpl(
-                Period.parse(streamSortSpecSpec.getWindowPeriod()),
-                streamSortSpecSpec.getWindowMargin(),
-                PartitionedEventTimeOrderingComparator.INSTANCE,
-                outputCollector);
-        this.streamSortSpecSpec = streamSortSpecSpec;
-        this.streamId = streamId;
-        this.outputCollector = outputCollector;
-    }
-
-    /**
-     * Entry point to manage window lifecycle
-     *
-     * @param event StreamEvent
-     */
-    public void nextEvent(PartitionedEvent event) {
-        final long eventTime = event.getEvent().getTimestamp();
-        boolean handled = false;
-
-        synchronized (this.windowManager){
-            for (StreamWindow window : this.windowManager.getWindows()) {
-                if (window.alive() && window.add(event)) {
-                    handled = true;
-                }
-            }
-
-            // No window found for the event but not too late being rejected
-            if (!handled && !windowManager.reject(eventTime)) {
-                // later then all events, create later window
-                StreamWindow window = windowManager.addNewWindow(eventTime);
-                if (window.add(event)) {
-                    LOG.info("Created {} of {} at {}", window, this.streamId, DateTimeUtil.millisecondsToHumanDateWithMilliseconds(eventTime));
-                    handled = true;
-                }
-            }
-        }
-
-        if(!handled){
-            if(LOG.isDebugEnabled()) {
-                LOG.debug("Drop expired event {}", event);
-            }
-            outputCollector.drop(event);
-        }
-    }
-
-    @Override
-    public void onTick(StreamTimeClock clock,long globalSystemTime) {
-        windowManager.onTick(clock, globalSystemTime);
-    }
-
-    @Override
-    public void close() {
-        try {
-            windowManager.close();
-        } catch (IOException e) {
-            LOG.error("Got exception while closing window manager",e);
-        }
-    }
-
-    @Override
-    public String toString() {
-        return super.toString();
-    }
-
-    @Override
-    public int hashCode(){
-        if(streamSortSpecSpec == null){
-            throw new NullPointerException("streamSortSpec is null");
-        }else{
-            return streamSortSpecSpec.hashCode();
-        }
-    }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/sorter/impl/StreamSortedWindowInMapDB.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/sorter/impl/StreamSortedWindowInMapDB.java b/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/sorter/impl/StreamSortedWindowInMapDB.java
deleted file mode 100644
index c1c289d..0000000
--- a/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/sorter/impl/StreamSortedWindowInMapDB.java
+++ /dev/null
@@ -1,145 +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.sorter.impl;
-
-import java.util.Arrays;
-import java.util.concurrent.atomic.AtomicInteger;
-
-import org.apache.commons.lang3.time.StopWatch;
-import org.apache.eagle.alert.engine.PartitionedEventCollector;
-import org.apache.eagle.alert.engine.model.PartitionedEvent;
-import org.apache.eagle.alert.engine.model.StreamEvent;
-import org.apache.eagle.alert.engine.sorter.BaseStreamWindow;
-import org.mapdb.BTreeMap;
-import org.mapdb.DB;
-import org.mapdb.Serializer;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * StreamSortedWindow based on MapDB to support off-heap or disk storage.
- *
- * Stable sorting algorithm
- *
- * <br/><br/>
- *
- * See <a href="http://www.mapdb.org">http://www.mapdb.org</a>
- */
-public class StreamSortedWindowInMapDB extends BaseStreamWindow {
-    private final String mapId;
-    private BTreeMap<Long, PartitionedEvent[]> bTreeMap;
-    private final static Logger LOG = LoggerFactory.getLogger(StreamSortedWindowInMapDB.class);
-    private final AtomicInteger size;
-    private  long replaceOpCount = 0;
-    private final static PartitionedEventGroupSerializer STREAM_EVENT_GROUP_SERIALIZER = new PartitionedEventGroupSerializer();
-
-    /**
-     * @param start
-     * @param end
-     * @param margin
-     * @param db
-     * @param mapId physical map id, used to decide whether to reuse or not
-     */
-    @SuppressWarnings("unused")
-    public StreamSortedWindowInMapDB(long start, long end, long margin,DB db,String mapId) {
-        super(start, end, margin);
-        this.mapId = mapId;
-        try {
-            bTreeMap = db.<Long, StreamEvent>treeMap(mapId).
-                    keySerializer(Serializer.LONG).
-                    valueSerializer(STREAM_EVENT_GROUP_SERIALIZER).
-                    createOrOpen();
-            LOG.debug("Created BTree map {}",mapId);
-        } catch (Error error){
-            LOG.info("Failed create BTree {}",mapId,error);
-        }
-        size = new AtomicInteger(0);
-    }
-
-    /**
-     * Assumed: most of adding operation will do putting only and few require replacing.
-     *
-     * <ol>
- *     <li>
-     * First of all, always try to put with created event directly
-     * </li>
-     * <li>
-     * If not absent (key already exists), then append and replace,
-     * replace operation will cause more consumption
-     * </li>
-     * </ol>
-     * @param event coming-in event
-     * @return whether success
-     */
-    @Override
-    public synchronized boolean add(PartitionedEvent event) {
-        long timestamp = event.getEvent().getTimestamp();
-        if(accept(timestamp)) {
-            boolean absent = bTreeMap.putIfAbsentBoolean(timestamp, new PartitionedEvent[]{event});
-            if (!absent) {
-                size.incrementAndGet();
-                return true;
-            } else {
-                if(LOG.isDebugEnabled()) LOG.debug("Duplicated timestamp {}, will reduce performance as replacing",timestamp);
-                PartitionedEvent[] oldValue = bTreeMap.get(timestamp);
-                PartitionedEvent[] newValue = oldValue == null ? new PartitionedEvent[1]: Arrays.copyOf(oldValue,oldValue.length+1);
-                newValue[newValue.length-1] = event;
-                PartitionedEvent[] removedValue = bTreeMap.replace(timestamp,newValue);
-                replaceOpCount ++;
-                if(replaceOpCount % 1000 == 0){
-                    LOG.warn("Too many events ({}) with overlap timestamp, may reduce insertion performance",replaceOpCount);
-                }
-                if(removedValue!=null) {
-                    size.incrementAndGet();
-                } else {
-                    throw new IllegalStateException("Failed to replace key "+timestamp+" with "+newValue.length+" entities array to replace old "+oldValue.length+" entities array");
-                }
-                return true;
-            }
-        } else {
-            return false;
-        }
-    }
-
-    @Override
-    protected synchronized void flush(PartitionedEventCollector collector) {
-        StopWatch stopWatch = new StopWatch();
-        stopWatch.start();
-        bTreeMap.valueIterator().forEachRemaining((events)->{
-            for(PartitionedEvent event:events){
-                collector.emit(event);
-            }
-        });
-        bTreeMap.clear();
-        replaceOpCount = 0;
-        stopWatch.stop();
-        LOG.info("Flushed {} events in {} ms", size, stopWatch.getTime());
-        size.set(0);
-    }
-
-    @Override
-    public synchronized void close(){
-        super.close();
-        bTreeMap.close();
-        LOG.info("Closed {}",this.mapId);
-    }
-
-    @Override
-    public synchronized int size() {
-        return size.get();
-    }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/sorter/impl/StreamSortedWindowOnHeap.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/sorter/impl/StreamSortedWindowOnHeap.java b/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/sorter/impl/StreamSortedWindowOnHeap.java
deleted file mode 100644
index d3b1d7d..0000000
--- a/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/sorter/impl/StreamSortedWindowOnHeap.java
+++ /dev/null
@@ -1,80 +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.sorter.impl;
-
-import java.util.Comparator;
-
-import org.apache.commons.lang3.time.StopWatch;
-import org.apache.eagle.alert.engine.PartitionedEventCollector;
-import org.apache.eagle.alert.engine.model.PartitionedEvent;
-import org.apache.eagle.alert.engine.sorter.BaseStreamWindow;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.google.common.collect.TreeMultiset;
-
-public class StreamSortedWindowOnHeap extends BaseStreamWindow {
-    private final static Logger LOG = LoggerFactory.getLogger(StreamSortedWindowOnHeap.class);
-    private final TreeMultiset<PartitionedEvent> treeMultisetCache;
-
-    /**
-     * @param start start time
-     * @param end end time
-     * @param margin margin time
-     */
-    public StreamSortedWindowOnHeap(long start, long end, long margin, Comparator<PartitionedEvent> comparator ){
-        super(start,end,margin);
-        treeMultisetCache = TreeMultiset.create(comparator);
-    }
-
-    public StreamSortedWindowOnHeap(long start, long end, long margin){
-        this(start,end,margin,new PartitionedEventTimeOrderingComparator());
-    }
-
-    @Override
-    public boolean add(PartitionedEvent partitionedEvent) {
-        synchronized (treeMultisetCache) {
-            if (accept(partitionedEvent.getEvent().getTimestamp())) {
-                treeMultisetCache.add(partitionedEvent);
-                return true;
-            } else {
-                if(LOG.isDebugEnabled()) LOG.debug("{} is not acceptable, ignored", partitionedEvent);
-                return false;
-            }
-        }
-    }
-
-    @Override
-    protected void flush(PartitionedEventCollector collector) {
-        synchronized (treeMultisetCache) {
-            StopWatch stopWatch = new StopWatch();
-            stopWatch.start();
-            treeMultisetCache.forEach(collector::emit);
-            int size = treeMultisetCache.size();
-            treeMultisetCache.clear();
-            stopWatch.stop();
-            LOG.info("Flushed {} events in {} ms from {}", size, stopWatch.getTime(),this.toString());
-        }
-    }
-
-    @Override
-    public int size() {
-        synchronized (treeMultisetCache) {
-            return treeMultisetCache.size();
-        }
-    }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/sorter/impl/StreamTimeClockInLocalMemory.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/sorter/impl/StreamTimeClockInLocalMemory.java b/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/sorter/impl/StreamTimeClockInLocalMemory.java
deleted file mode 100644
index 91a4a37..0000000
--- a/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/sorter/impl/StreamTimeClockInLocalMemory.java
+++ /dev/null
@@ -1,64 +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.sorter.impl;
-
-import java.util.concurrent.atomic.AtomicLong;
-
-import org.apache.eagle.alert.engine.sorter.StreamTimeClock;
-import org.apache.eagle.alert.utils.DateTimeUtil;
-
-
-/**
- * In memory thread-safe time clock service
- *
- * TODO: maybe need to synchronize time clock globally, how to?
- */
-public class StreamTimeClockInLocalMemory implements StreamTimeClock {
-    private final AtomicLong currentTime;
-    private final String streamId;
-
-    public StreamTimeClockInLocalMemory(String streamId,long initialTime){
-        this.streamId = streamId;
-        this.currentTime = new AtomicLong(initialTime);
-    }
-    public StreamTimeClockInLocalMemory(String streamId){
-        this(streamId,0L);
-    }
-
-    @Override
-    public void moveForward(long timestamp){
-        if(timestamp < currentTime.get()){
-            throw new IllegalArgumentException(timestamp +" < "+currentTime.get()+", should not move time back");
-        }
-        this.currentTime.set(timestamp);
-    }
-
-    @Override
-    public String getStreamId() {
-        return streamId;
-    }
-
-    @Override
-    public long getTime() {
-        return currentTime.get();
-    }
-
-    @Override
-    public String toString() {
-        return String.format("StreamClock[streamId=%s, now=%s]",streamId, DateTimeUtil.millisecondsToHumanDateWithMilliseconds(currentTime.get()));
-    }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/sorter/impl/StreamTimeClockManagerImpl.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/sorter/impl/StreamTimeClockManagerImpl.java b/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/sorter/impl/StreamTimeClockManagerImpl.java
deleted file mode 100644
index 741d7f0..0000000
--- a/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/sorter/impl/StreamTimeClockManagerImpl.java
+++ /dev/null
@@ -1,171 +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.sorter.impl;
-
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Timer;
-import java.util.TimerTask;
-import java.util.concurrent.atomic.AtomicInteger;
-
-import org.apache.eagle.alert.engine.sorter.StreamTimeClock;
-import org.apache.eagle.alert.engine.sorter.StreamTimeClockListener;
-import org.apache.eagle.alert.engine.sorter.StreamTimeClockManager;
-import org.apache.eagle.alert.utils.DateTimeUtil;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public final class StreamTimeClockManagerImpl implements StreamTimeClockManager {
-    private static final long serialVersionUID = -2770823821511195343L;
-    private final static Logger LOG = LoggerFactory.getLogger(StreamTimeClockManagerImpl.class);
-    private final Map<String,StreamTimeClock> streamIdTimeClockMap;
-    private Timer timer;
-
-    private final Map<StreamTimeClockListener,String> listenerStreamIdMap;
-    private final static AtomicInteger num = new AtomicInteger();
-
-    public StreamTimeClockManagerImpl(){
-        listenerStreamIdMap = new HashMap<>();
-        streamIdTimeClockMap = new HashMap<>();
-        timer = new Timer("StreamScheduler-"+num.getAndIncrement());
-        timer.schedule(new TimerTask() {
-            @Override
-            public void run() {
-                // Make sure the timer tick happens one by one
-                    triggerTickOnAll();
-            }
-        },1000,1000);
-    }
-
-    /**
-     *
-     * By default, we could keep the current time clock in memory,
-     * Eventually we may need to consider the global time synchronization across all nodes
-     *
-     * 1) When to initialize window according to start time
-     * 2) When to close expired window according to current time
-     *
-     * @return StreamTimeClock instance
-     */
-    @Override
-    public StreamTimeClock createStreamTimeClock(String streamId){
-        synchronized (streamIdTimeClockMap) {
-            if (!streamIdTimeClockMap.containsKey(streamId)) {
-                StreamTimeClock instance = new StreamTimeClockInLocalMemory(streamId);
-                LOG.info("Created {}", instance);
-                streamIdTimeClockMap.put(streamId, instance);
-            } else {
-                LOG.warn("TimeClock for stream already existss: "+streamIdTimeClockMap.get(streamId));
-            }
-            return streamIdTimeClockMap.get(streamId);
-        }
-    }
-
-    /**
-     * @param streamId
-     * @return
-     */
-    @Override
-    public StreamTimeClock getStreamTimeClock(String streamId) {
-        synchronized (streamIdTimeClockMap) {
-            if (!streamIdTimeClockMap.containsKey(streamId)) {
-                LOG.warn("TimeClock for stream {} is not initialized before being called, create now", streamId);
-                return createStreamTimeClock(streamId);
-            }
-            return streamIdTimeClockMap.get(streamId);
-        }
-    }
-
-    /**
-     * @param streamId
-     */
-    @Override
-    public void removeStreamTimeClock(String streamId){
-        synchronized (streamIdTimeClockMap) {
-            if (streamIdTimeClockMap.containsKey(streamId)) {
-                streamIdTimeClockMap.remove(streamId);
-                LOG.info("Removed TimeClock for stream {}: {}", streamId, streamIdTimeClockMap.get(streamId));
-            } else {
-                LOG.warn("No TimeClock found for stream {}, nothing to remove", streamId);
-            }
-        }
-    }
-
-    @Override
-    public void registerListener(String streamId,StreamTimeClockListener listener) {
-        synchronized (listenerStreamIdMap) {
-            if (listenerStreamIdMap.containsKey(listener))
-                throw new IllegalArgumentException("Duplicated listener: " + listener.toString());
-            LOG.info("Register {} on {}", listener, streamId);
-            listenerStreamIdMap.put(listener, streamId);
-        }
-    }
-
-    @Override
-    public void registerListener(StreamTimeClock streamClock, StreamTimeClockListener listener) {
-        registerListener(streamClock.getStreamId(),listener);
-    }
-
-    @Override
-    public void removeListener(StreamTimeClockListener listener) {
-        listenerStreamIdMap.remove(listener);
-    }
-
-    @Override
-    public synchronized void triggerTickOn(String streamId) {
-        int count = 0;
-        for (Map.Entry<StreamTimeClockListener, String> entry : listenerStreamIdMap.entrySet()) {
-            if (entry.getValue().equals(streamId)) {
-                entry.getKey().onTick(streamIdTimeClockMap.get(streamId), getCurrentSystemTime());
-                count++;
-            }
-        }
-        if (LOG.isDebugEnabled()) LOG.debug("Triggered {} time-clock listeners on stream {}", count, streamId);
-    }
-
-    private static long getCurrentSystemTime(){
-        return System.currentTimeMillis();
-    }
-
-    @Override
-    public void onTimeUpdate(String streamId, long timestamp) {
-        StreamTimeClock timeClock = getStreamTimeClock(streamId);
-        if(timeClock == null)
-            return;
-        // Trigger time clock only when time moves forward
-        if(timestamp >= timeClock.getTime()) {
-            timeClock.moveForward(timestamp);
-            if(LOG.isDebugEnabled()) LOG.debug("Tick on stream {} with latest time {}",streamId, DateTimeUtil.millisecondsToHumanDateWithMilliseconds(timeClock.getTime()));
-            triggerTickOn(streamId);
-        }
-    }
-
-    private void triggerTickOnAll(){
-        synchronized (listenerStreamIdMap) {
-            for (Map.Entry<StreamTimeClockListener, String> entry : listenerStreamIdMap.entrySet()) {
-                triggerTickOn(entry.getValue());
-            }
-        }
-    }
-
-    @Override
-    public void close() {
-        timer.cancel();
-        triggerTickOnAll();
-        LOG.info("Closed StreamTimeClockManager {}",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-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/sorter/impl/StreamWindowManagerImpl.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/sorter/impl/StreamWindowManagerImpl.java b/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/sorter/impl/StreamWindowManagerImpl.java
deleted file mode 100644
index 4e1212f..0000000
--- a/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/sorter/impl/StreamWindowManagerImpl.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
- * <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.sorter.impl;
-
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Comparator;
-import java.util.List;
-import java.util.TreeMap;
-
-import org.apache.commons.lang3.time.StopWatch;
-import org.apache.eagle.alert.engine.PartitionedEventCollector;
-import org.apache.eagle.alert.engine.model.PartitionedEvent;
-import org.apache.eagle.alert.engine.sorter.StreamTimeClock;
-import org.apache.eagle.alert.engine.sorter.StreamWindow;
-import org.apache.eagle.alert.engine.sorter.StreamWindowManager;
-import org.apache.eagle.alert.engine.sorter.StreamWindowRepository;
-import org.apache.eagle.alert.utils.DateTimeUtil;
-import org.apache.eagle.alert.utils.TimePeriodUtils;
-import org.joda.time.Period;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class StreamWindowManagerImpl implements StreamWindowManager {
-    private final static Logger LOG = LoggerFactory.getLogger(StreamWindowManagerImpl.class);
-    private final TreeMap<Long,StreamWindow> windowBuckets;
-    private final PartitionedEventCollector collector;
-    private final Period windowPeriod;
-    private final long windowMargin;
-    @SuppressWarnings("unused")
-    private final Comparator<PartitionedEvent> comparator;
-    private long rejectTime;
-
-    public StreamWindowManagerImpl(Period windowPeriod, long windowMargin, Comparator<PartitionedEvent> comparator, PartitionedEventCollector collector){
-        this.windowBuckets = new TreeMap<>();
-        this.windowPeriod = windowPeriod;
-        this.windowMargin = windowMargin;
-        this.collector = collector;
-        this.comparator = comparator;
-    }
-
-    @Override
-    public StreamWindow addNewWindow(long initialTime) {
-        synchronized (windowBuckets) {
-            if (!reject(initialTime)) {
-                Long windowStartTime = TimePeriodUtils.formatMillisecondsByPeriod(initialTime, windowPeriod);
-                Long windowEndTime = windowStartTime + TimePeriodUtils.getMillisecondsOfPeriod(windowPeriod);
-                StreamWindow window = StreamWindowRepository.getSingletonInstance().createWindow(windowStartTime, windowEndTime, windowMargin);
-                window.register(collector);
-                addWindow(window);
-                return window;
-            } else {
-                throw new IllegalStateException("Failed to create new window, as " + DateTimeUtil.millisecondsToHumanDateWithMilliseconds(initialTime) + " is too late, only allow timestamp after " + DateTimeUtil.millisecondsToHumanDateWithMilliseconds(rejectTime));
-            }
-        }
-    }
-
-    private void addWindow(StreamWindow window){
-        if (!windowBuckets.containsKey(window.startTime())) {
-            windowBuckets.put(window.startTime(), window);
-        } else {
-            throw new IllegalArgumentException("Duplicated " + window.toString());
-        }
-    }
-
-    @Override
-    public void removeWindow(StreamWindow window) {
-        synchronized (windowBuckets) {
-            windowBuckets.remove(window.startTime());
-        }
-    }
-
-    @Override
-    public boolean hasWindow(StreamWindow window) {
-        synchronized (windowBuckets) {
-            return windowBuckets.containsKey(window.startTime());
-        }
-    }
-
-    @Override
-    public boolean hasWindowFor(long timestamp) {
-        return getWindowFor(timestamp) != null;
-    }
-
-    @Override
-    public Collection<StreamWindow> getWindows() {
-        synchronized (windowBuckets) {
-            return windowBuckets.values();
-        }
-    }
-
-    @Override
-    public StreamWindow getWindowFor(long timestamp) {
-        synchronized (windowBuckets) {
-            for (StreamWindow windowBucket : windowBuckets.values()) {
-                if (timestamp >= windowBucket.startTime() && timestamp < windowBucket.endTime()) {
-                    return windowBucket;
-                }
-            }
-            return null;
-        }
-    }
-
-    @Override
-    public boolean reject(long timestamp) {
-        return timestamp < rejectTime;
-    }
-
-    @Override
-    public void onTick(StreamTimeClock clock,long globalSystemTime) {
-        synchronized (windowBuckets) {
-            List<StreamWindow> toRemoved = new ArrayList<>();
-            List<StreamWindow> aliveWindow = new ArrayList<>();
-
-            for (StreamWindow windowBucket : windowBuckets.values()) {
-                windowBucket.onTick(clock, globalSystemTime);
-                if (windowBucket.rejectTime() > rejectTime) rejectTime = windowBucket.rejectTime();
-            }
-            for (StreamWindow windowBucket : windowBuckets.values()) {
-                if (windowBucket.expired() || windowBucket.endTime() <=rejectTime) {
-                    toRemoved.add(windowBucket);
-                } else {
-                    aliveWindow.add(windowBucket);
-                }
-            }
-            toRemoved.forEach(this::CloseAndRemoveWindow);
-            if (toRemoved.size() > 0) LOG.info("Windows: {} alive = {}, {} expired = {}", aliveWindow.size(), aliveWindow, toRemoved.size(), toRemoved);
-        }
-    }
-
-    private void CloseAndRemoveWindow(StreamWindow windowBucket){
-        StopWatch stopWatch = new StopWatch();
-        stopWatch.start();
-        CloseWindow(windowBucket);
-        removeWindow(windowBucket);
-        stopWatch.stop();
-        LOG.info("Removed {} in {} ms",windowBucket,stopWatch.getTime());
-    }
-
-    private void CloseWindow(StreamWindow windowBucket){
-        windowBucket.close();
-    }
-
-    public void close() {
-        synchronized (windowBuckets) {
-            LOG.debug("Closing");
-            StopWatch stopWatch = new StopWatch();
-            stopWatch.start();
-            int count = 0;
-            for (StreamWindow windowBucket : getWindows()) {
-                count++;
-                CloseWindow(windowBucket);
-            }
-            windowBuckets.clear();
-            stopWatch.stop();
-            LOG.info("Closed {} windows in {} ms", count, stopWatch.getTime());
-        }
-    }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/spout/CorrelationSpout.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/spout/CorrelationSpout.java b/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/spout/CorrelationSpout.java
deleted file mode 100644
index cd23405..0000000
--- a/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/spout/CorrelationSpout.java
+++ /dev/null
@@ -1,380 +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.spout;
-
-import java.text.MessageFormat;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-import org.apache.commons.collections.CollectionUtils;
-import org.apache.eagle.alert.coordination.model.Kafka2TupleMetadata;
-import org.apache.eagle.alert.coordination.model.SpoutSpec;
-import org.apache.eagle.alert.engine.coordinator.IMetadataChangeNotifyService;
-import org.apache.eagle.alert.engine.coordinator.MetadataType;
-import org.apache.eagle.alert.engine.coordinator.StreamDefinition;
-import org.apache.eagle.alert.engine.router.SpoutSpecListener;
-import org.apache.eagle.alert.engine.serialization.PartitionedEventSerializer;
-import org.apache.eagle.alert.engine.serialization.SerializationMetadataProvider;
-import org.apache.eagle.alert.engine.serialization.Serializers;
-import org.apache.eagle.alert.utils.AlertConstants;
-import org.apache.eagle.alert.utils.StreamIdConversion;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import storm.kafka.BrokerHosts;
-import storm.kafka.KafkaSpoutMetric;
-import storm.kafka.KafkaSpoutWrapper;
-import storm.kafka.SpoutConfig;
-import storm.kafka.ZkHosts;
-import backtype.storm.spout.SchemeAsMultiScheme;
-import backtype.storm.spout.SpoutOutputCollector;
-import backtype.storm.task.TopologyContext;
-import backtype.storm.topology.OutputFieldsDeclarer;
-import backtype.storm.topology.base.BaseRichSpout;
-import backtype.storm.tuple.Fields;
-
-import com.typesafe.config.Config;
-
-/**
- * wrap KafkaSpout to provide parallel processing of messages for multiple Kafka topics
- *
- * 1. onNewConfig() is interface for outside to update new metadata. Upon new metadata, this class will calculate if there is any new topic, removed topic or
- *    updated topic
- *
- */
-public class CorrelationSpout extends BaseRichSpout implements SpoutSpecListener,SerializationMetadataProvider {
-    private static final long serialVersionUID = -5280723341236671580L;
-    private static final Logger LOG  = LoggerFactory.getLogger(CorrelationSpout.class);
-
-    public static final String DEFAULT_STORM_KAFKA_TRANSACTION_ZK_ROOT = "/consumers";
-    public static final String DEFAULT_STORM_KAFKA_TRANSACTION_ZK_RELATIVE_PATH = "/eagle_consumer";
-
-    // topic to KafkaSpoutWrapper
-    private volatile Map<String, KafkaSpoutWrapper> kafkaSpoutList = new HashMap<>();
-    private int numOfRouterBolts;
-
-    private SpoutSpec cachedSpoutSpec;
-
-    private transient KafkaSpoutMetric kafkaSpoutMetric;
-
-    @SuppressWarnings("rawtypes")
-    private Map conf;
-    private TopologyContext context;
-    private SpoutOutputCollector collector;
-    private final Config config;
-    private String topologyId;
-    private String spoutName;
-    private String routeBoltName;
-    @SuppressWarnings("unused")
-    private int taskIndex;
-    private IMetadataChangeNotifyService changeNotifyService;
-    private PartitionedEventSerializer serializer;
-    private volatile Map<String, StreamDefinition> sds;
-
-    /**
-     * FIXME one single changeNotifyService may have issues as possibly multiple spout tasks will register themselves and initialize service
-     * @param config
-     * @param topologyId
-     * @param changeNotifyService
-     * @param numOfRouterBolts
-     */
-    public CorrelationSpout(Config config, String topologyId, IMetadataChangeNotifyService changeNotifyService, int numOfRouterBolts){
-        this(config, topologyId, changeNotifyService, numOfRouterBolts, AlertConstants.DEFAULT_SPOUT_NAME, AlertConstants.DEFAULT_ROUTERBOLT_NAME);
-    }
-    /**
-     *
-     * @param config
-     * @param topologyId used for distinguishing kafka offset for different topologies
-     * @param numOfRouterBolts used for generating streamId and routing
-     * @param spoutName used for generating streamId between spout and router bolt
-     * @param routerBoltName used for generating streamId between spout and router bolt
-     */
-    public CorrelationSpout(Config config, String topologyId, IMetadataChangeNotifyService changeNotifyService, int numOfRouterBolts, String spoutName, String routerBoltName){
-        this.config = config;
-        this.topologyId = topologyId;
-        this.changeNotifyService = changeNotifyService;
-        this.numOfRouterBolts = numOfRouterBolts;
-        this.spoutName = spoutName;
-        this.routeBoltName = routerBoltName;
-    }
-
-    public String getSpoutName(){
-        return spoutName;
-    }
-
-    public String getRouteBoltName(){
-        return routeBoltName;
-    }
-
-    /**
-     * the only output field is for StreamEvent
-     * @param declarer
-     */
-    @Override
-    public void declareOutputFields(OutputFieldsDeclarer declarer) {
-        for (int i = 0; i < numOfRouterBolts; i++) {
-            String streamId = StreamIdConversion.generateStreamIdBetween(spoutName, routeBoltName + i);
-            declarer.declareStream(streamId, new Fields(AlertConstants.FIELD_0));
-            LOG.info("declare stream between spout and streamRouterBolt " + streamId);
-        }
-    }
-
-    @SuppressWarnings("rawtypes")
-    @Override
-    public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
-        if(LOG.isDebugEnabled()) {
-            LOG.debug("open method invoked");
-        }
-        this.conf = conf;
-        this.context = context;
-        this.collector = collector;
-        this.taskIndex = context.getThisTaskIndex();
-
-        // initialize an empty SpoutSpec
-        cachedSpoutSpec = new SpoutSpec(topologyId, new HashMap<>(), new HashMap<>(), new HashMap<>());
-
-        changeNotifyService.registerListener(this);
-        changeNotifyService.init(config, MetadataType.SPOUT);
-
-        // register KafkaSpout metric
-        kafkaSpoutMetric = new KafkaSpoutMetric();
-        context.registerMetric("kafkaSpout", kafkaSpoutMetric, 60);
-
-        this.serializer = Serializers.newPartitionedEventSerializer(this);
-    }
-
-    @Override
-    public void onSpoutSpecChange(SpoutSpec spec, Map<String, StreamDefinition> sds) {
-        LOG.info("new metadata is updated " + spec);
-        try{
-            onReload(spec, sds);
-        }catch(Exception ex){
-            LOG.error("error applying new SpoutSpec", ex);
-        }
-    }
-
-    @Override
-    public void nextTuple() {
-        for (KafkaSpoutWrapper wrapper : kafkaSpoutList.values()) {
-            wrapper.nextTuple();
-        }
-    }
-
-    /**
-     * find the correct wrapper to do ack that means msgId should be mapped to
-     * wrapper
-     *
-     * @param msgId
-     */
-    @Override
-    public void ack(Object msgId) {
-        // decode and get topic
-        KafkaMessageIdWrapper id = (KafkaMessageIdWrapper) msgId;
-        KafkaSpoutWrapper spout = kafkaSpoutList.get(id.topic);
-        spout.ack(id.id);
-    }
-
-    @Override
-    public void fail(Object msgId) {
-        // decode and get topic
-        KafkaMessageIdWrapper id = (KafkaMessageIdWrapper) msgId;
-        LOG.error("Failing message {}, with topic {}", msgId, id.topic);
-        KafkaSpoutWrapper spout = kafkaSpoutList.get(id.topic);
-        spout.fail(id.id);
-    }
-
-    @Override
-    public void deactivate() {
-        System.out.println("deactivate");
-        for (KafkaSpoutWrapper wrapper : kafkaSpoutList.values()) {
-            wrapper.deactivate();
-        }
-    }
-
-    @Override
-    public void close() {
-        System.out.println("close");
-        for (KafkaSpoutWrapper wrapper : kafkaSpoutList.values()) {
-            wrapper.close();
-        }
-    }
-    
-    private List<String> getTopics(SpoutSpec spoutSpec) {
-        List<String> meta = new ArrayList<String>();
-        for (Kafka2TupleMetadata entry : spoutSpec.getKafka2TupleMetadataMap().values()) {
-            meta.add(entry.getTopic());
-        }
-        return meta;
-    }
-
-    @SuppressWarnings("unchecked")
-    public void onReload(final SpoutSpec newMeta, Map<String, StreamDefinition> sds) throws Exception {
-        // calculate topic create/remove/update
-        List<String> topics = getTopics(newMeta);
-        List<String> cachedTopcies = getTopics(cachedSpoutSpec);
-        Collection<String> newTopics = CollectionUtils.subtract(topics, cachedTopcies);
-        Collection<String> removeTopics = CollectionUtils.subtract(cachedTopcies, topics);
-        Collection<String> updateTopics = CollectionUtils.intersection(topics, cachedTopcies);
-
-        LOG.info("Topics were added={}, removed={}, modified={}", newTopics, removeTopics, updateTopics);
-
-        // build lookup table for scheme
-        Map<String, String> newSchemaName = new HashMap<String, String>();
-        for (Kafka2TupleMetadata ds : newMeta.getKafka2TupleMetadataMap().values()) {
-            newSchemaName.put(ds.getTopic(), ds.getSchemeCls());
-        }
-
-        // copy and swap
-        Map<String, KafkaSpoutWrapper> newKafkaSpoutList = new HashMap<>(this.kafkaSpoutList);
-        // iterate new topics and then create KafkaSpout
-        for(String topic : newTopics){
-            KafkaSpoutWrapper wrapper = newKafkaSpoutList.get(topic);
-            if (wrapper != null) {
-                LOG.warn(MessageFormat.format("try to create new topic {0}, but found in the active spout list, this may indicate some inconsistency", topic));
-                continue;
-            }
-            KafkaSpoutWrapper newWrapper = createKafkaSpout(conf, context, collector, topic, newSchemaName.get(topic), newMeta, sds);
-            newKafkaSpoutList.put(topic, newWrapper);
-        }
-        // iterate remove topics and then close KafkaSpout
-        for(String topic : removeTopics){
-            KafkaSpoutWrapper wrapper = newKafkaSpoutList.get(topic);
-            if (wrapper == null) {
-                LOG.warn(MessageFormat.format("try to remove topic {0}, but not found in the active spout list, this may indicate some inconsistency", topic));
-                continue;
-            }
-            removeKafkaSpout(wrapper);
-            newKafkaSpoutList.remove(topic);
-        }
-
-        // iterate update topic and then update metadata
-        for(String topic : updateTopics){
-            KafkaSpoutWrapper spoutWrapper = newKafkaSpoutList.get(topic);
-            if (spoutWrapper == null) {
-                LOG.warn(MessageFormat.format("try to update topic {0}, but not found in the active spout list, this may indicate some inconsistency", topic));
-                continue;
-            }
-            spoutWrapper.update(newMeta, sds);
-        }
-
-        // swap
-        this.cachedSpoutSpec = newMeta;
-        this.kafkaSpoutList = newKafkaSpoutList;
-        this.sds = sds;
-    }
-
-    /**
-     * make this method protected to make sure unit test can work well
-     * Q: Where to persist consumer state, i.e. what offset has been consumed for each topic and partition
-     * A: stormKafkaTransactionZkPath + "/" + consumerId + "/" + topic + "/" + topologyId + "/" + partitionId
-     * Note1: PartitionManager.committedPath for composing zkState path,  _spoutConfig.zkRoot + "/" + _spoutConfig.id + "/" + _partition.getId();
-     * consumerId by default is EagleConsumer unless it is specified by "stormKafkaEagleConsumer"
-     * Note2: put topologyId as part of zkState because one topic by design can be consumed by multiple topologies so one topology needs to know
-     * processed offset for itself
-     *
-     * TODO: Should avoid use Config.get in deep calling stack, should generate config bean as early as possible
-     *
-     * @param conf
-     * @param context
-     * @param collector
-     * @param topic
-     * @param spoutSpec
-     * @return
-     */
-    @SuppressWarnings("rawtypes")
-    protected KafkaSpoutWrapper createKafkaSpout(Map conf, TopologyContext context, SpoutOutputCollector collector, final String topic,
-                                                 String schemeClsName, SpoutSpec spoutSpec, Map<String, StreamDefinition> sds) throws Exception{
-        String kafkaBrokerZkQuorum = config.getString("spout.kafkaBrokerZkQuorum");
-        BrokerHosts hosts = new ZkHosts(kafkaBrokerZkQuorum);
-        String transactionZkRoot = DEFAULT_STORM_KAFKA_TRANSACTION_ZK_ROOT;
-        if(config.hasPath("spout.stormKafkaTransactionZkPath")) {
-            transactionZkRoot = config.getString("spout.stormKafkaTransactionZkPath");
-        }
-        // write partition offset etc. into zkRoot+id, see PartitionManager.committedPath
-        String zkStateTransactionRelPath = DEFAULT_STORM_KAFKA_TRANSACTION_ZK_RELATIVE_PATH;
-        if(config.hasPath("spout.stormKafkaEagleConsumer")){
-            zkStateTransactionRelPath = config.getString("spout.stormKafkaEagleConsumer");
-        }
-        SpoutConfig spoutConfig = new SpoutConfig(hosts, topic, transactionZkRoot, zkStateTransactionRelPath + "/" + topic + "/" + topologyId);
-        // transaction zkServers
-        boolean stormKafkaUseSameZkQuorumWithKafkaBroker = config.getBoolean("spout.stormKafkaUseSameZkQuorumWithKafkaBroker");
-        if(stormKafkaUseSameZkQuorumWithKafkaBroker){
-            ZkServerPortUtils utils = new ZkServerPortUtils(kafkaBrokerZkQuorum);
-            spoutConfig.zkServers = utils.getZkHosts();
-            spoutConfig.zkPort = utils.getZkPort();
-        }else{
-            ZkServerPortUtils utils = new ZkServerPortUtils(config.getString("spout.stormKafkaTransactionZkQuorum"));
-            spoutConfig.zkServers = utils.getZkHosts();
-            spoutConfig.zkPort = utils.getZkPort();
-        }
-        // transaction update interval
-        spoutConfig.stateUpdateIntervalMs = config.getLong("spout.stormKafkaStateUpdateIntervalMs");
-        // Kafka fetch size
-        spoutConfig.fetchSizeBytes = config.getInt("spout.stormKafkaFetchSizeBytes");
-        // "startOffsetTime" is for test usage, prod should not use this
-        if (config.hasPath("spout.stormKafkaStartOffsetTime")) {
-            spoutConfig.startOffsetTime = config.getInt("spout.stormKafkaStartOffsetTime");
-        }
-
-        spoutConfig.scheme = new SchemeAsMultiScheme(SchemeBuilder.buildFromClsName(schemeClsName, topic));
-        KafkaSpoutWrapper wrapper = new KafkaSpoutWrapper(spoutConfig, kafkaSpoutMetric);
-        SpoutOutputCollectorWrapper collectorWrapper = new SpoutOutputCollectorWrapper(this, collector, topic, spoutSpec, numOfRouterBolts, sds,this.serializer);
-        wrapper.open(conf, context, collectorWrapper);
-        return wrapper;
-    }
-
-    @Override
-    public StreamDefinition getStreamDefinition(String streamId) {
-        return sds.get(streamId);
-    }
-
-    /**
-     * utility to get list of zkServers and zkPort.(It is assumed that zkPort is same for all zkServers as storm-kafka library requires this though it is not efficient)
-     */
-    private static class ZkServerPortUtils{
-        private List<String> zkHosts = new ArrayList<>();
-        private Integer zkPort;
-        public ZkServerPortUtils(String zkQuorum){
-            String[] zkConnections = zkQuorum.split(",");
-            for (String zkConnection : zkConnections) {
-                zkHosts.add(zkConnection.split(":")[0]);
-            }
-            zkPort = Integer.valueOf(zkConnections[0].split(":")[1]);
-        }
-
-        public List<String> getZkHosts(){
-            return zkHosts;
-        }
-
-        public Integer getZkPort(){
-            return zkPort;
-        }
-    }
-
-    protected void removeKafkaSpout(KafkaSpoutWrapper wrapper){
-        try {
-            wrapper.close();
-        } catch (Exception e) {
-            LOG.error("Close wrapper failed. Ignore and continue!", e);
-        }
-    }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/spout/CreateTopicUtils.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/spout/CreateTopicUtils.java b/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/spout/CreateTopicUtils.java
deleted file mode 100644
index a2c9219..0000000
--- a/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/spout/CreateTopicUtils.java
+++ /dev/null
@@ -1,43 +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.spout;
-
-import java.util.Properties;
-
-import kafka.admin.AdminUtils;
-import kafka.utils.ZKStringSerializer$;
-
-import org.I0Itec.zkclient.ZkClient;
-import org.slf4j.Logger;
-
-/**
- * normally this is used in unit test for convenience
- */
-public class CreateTopicUtils {
-    private static final Logger LOG = org.slf4j.LoggerFactory.getLogger(CreateTopicUtils.class);
-    private static final int partitions = 2;
-    private static final int replicationFactor = 1;
-    public static void ensureTopicReady(String zkQuorum, String topic){
-        ZkClient zkClient = new ZkClient(zkQuorum, 10000, 10000, ZKStringSerializer$.MODULE$);
-        if(!AdminUtils.topicExists(zkClient, topic)) {
-            LOG.info("create topic " + topic + " with partitions " + partitions + ", and replicationFactor " + replicationFactor);
-            AdminUtils.createTopic(zkClient, topic, partitions, replicationFactor, new Properties());
-        }
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/spout/ISpoutSpecLCM.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/spout/ISpoutSpecLCM.java b/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/spout/ISpoutSpecLCM.java
deleted file mode 100644
index 23e94c3..0000000
--- a/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/spout/ISpoutSpecLCM.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.spout;
-
-import java.util.Map;
-
-import org.apache.eagle.alert.coordination.model.SpoutSpec;
-import org.apache.eagle.alert.engine.coordinator.StreamDefinition;
-
-/**
- * topic to stream metadata lifecycle method
- * one topic may spawn multiple streams, the metadata change includes
- * 1. add/remove stream
- * 2. for a specific stream, groupingstrategy is changed
- *    ex1, this stream has more alert bolts than before, then this spout would take more traffic
- *    ex2, this stream has less alert bolts than before, then this spout would take less traffic
- */
-public interface ISpoutSpecLCM {
-    /**
-     * stream metadata is used for SPOUT to filter traffic and route traffic to following groupby bolts.
-     * @param metadata
-     */
-    void update(SpoutSpec metadata, Map<String, StreamDefinition> sds);
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/spout/KafkaMessageIdWrapper.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/spout/KafkaMessageIdWrapper.java b/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/spout/KafkaMessageIdWrapper.java
deleted file mode 100644
index c786c01..0000000
--- a/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/spout/KafkaMessageIdWrapper.java
+++ /dev/null
@@ -1,42 +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.spout;
-
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.databind.ObjectMapper;
-
-/**
- * Created on 2/18/16.
- */
-public class KafkaMessageIdWrapper {
-    public Object id;
-    public KafkaMessageIdWrapper(Object o){
-        this.id = o;
-    }
-    public String topic;
-    private final static ObjectMapper objectMapper = new ObjectMapper();
-
-    public String toString(){
-        try {
-            return String.format("KafkaMessageIdWrapper[topic=%s, id=%s]", topic, objectMapper.writeValueAsString(id));
-        } catch (JsonProcessingException e) {
-            throw new IllegalStateException(e);
-        }
-    }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/spout/SchemeBuilder.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/spout/SchemeBuilder.java b/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/spout/SchemeBuilder.java
deleted file mode 100644
index 223f1b5..0000000
--- a/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/spout/SchemeBuilder.java
+++ /dev/null
@@ -1,35 +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.spout;
-
-import backtype.storm.spout.Scheme;
-
-
-/**
- * All Scheme implementations should have the following conditions
- * 1) implement Scheme interface
- * 2) has one constructor with topic name as parameter
- */
-public class SchemeBuilder {
-    public static Scheme buildFromClsName(String clsName, String topic) throws Exception{
-        Object o = Class.forName(clsName).getConstructor(String.class).newInstance(topic);
-        return (Scheme)o;
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/spout/SpoutOutputCollectorWrapper.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/spout/SpoutOutputCollectorWrapper.java b/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/spout/SpoutOutputCollectorWrapper.java
deleted file mode 100644
index b37f7b3..0000000
--- a/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/spout/SpoutOutputCollectorWrapper.java
+++ /dev/null
@@ -1,220 +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.spout;
-
-import java.io.IOException;
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
-
-import org.apache.commons.lang3.builder.HashCodeBuilder;
-import org.apache.eagle.alert.coordination.model.SpoutSpec;
-import org.apache.eagle.alert.coordination.model.StreamRepartitionMetadata;
-import org.apache.eagle.alert.coordination.model.StreamRepartitionStrategy;
-import org.apache.eagle.alert.coordination.model.Tuple2StreamConverter;
-import org.apache.eagle.alert.engine.coordinator.StreamDefinition;
-import org.apache.eagle.alert.engine.coordinator.StreamPartition;
-import org.apache.eagle.alert.engine.model.PartitionedEvent;
-import org.apache.eagle.alert.engine.model.StreamEvent;
-import org.apache.eagle.alert.engine.serialization.PartitionedEventSerializer;
-import org.apache.eagle.alert.engine.serialization.SerializationMetadataProvider;
-import org.apache.eagle.alert.utils.StreamIdConversion;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import backtype.storm.spout.ISpoutOutputCollector;
-import backtype.storm.spout.SpoutOutputCollector;
-
-/**
- * intercept the message sent from within KafkaSpout and select downstream bolts based on meta-data
- * This is topic based. each topic will have one SpoutOutputCollectorWrapper
- */
-public class SpoutOutputCollectorWrapper extends SpoutOutputCollector implements ISpoutSpecLCM,SerializationMetadataProvider {
-    private static final Logger LOG = LoggerFactory.getLogger(SpoutOutputCollectorWrapper.class);
-
-    private final ISpoutOutputCollector delegate;
-    private final String topic;
-    private final PartitionedEventSerializer serializer;
-    private int numOfRouterBolts;
-
-    private volatile List<StreamRepartitionMetadata> streamRepartitionMetadataList;
-    private volatile Tuple2StreamConverter converter;
-    private CorrelationSpout spout;
-    private volatile Map<String, StreamDefinition> sds;
-
-    /**
-     * @param delegate   actual SpoutOutputCollector to send data to following bolts
-     * @param topic      topic for this KafkaSpout to handle
-     * @param numGroupbyBolts bolts following this spout
-     * @param serializer
-     */
-    public SpoutOutputCollectorWrapper(CorrelationSpout spout,
-                                       ISpoutOutputCollector delegate,
-                                       String topic,
-                                       SpoutSpec spoutSpec,
-                                       int numGroupbyBolts,
-                                       Map<String, StreamDefinition> sds, PartitionedEventSerializer serializer) {
-        super(delegate);
-        this.spout = spout;
-        this.delegate = delegate;
-        this.topic = topic;
-        this.streamRepartitionMetadataList = spoutSpec.getStreamRepartitionMetadataMap().get(topic);
-        this.converter = new Tuple2StreamConverter(spoutSpec.getTuple2StreamMetadataMap().get(topic));
-        this.numOfRouterBolts = numGroupbyBolts;
-        this.sds = sds;
-        this.serializer = serializer;
-    }
-
-    /**
-     * How to assert that numTotalGroupbyBolts >= numOfRouterBolts, otherwise
-     * there is runtime issue by default, tuple includes 2 fields field 1: topic
-     * name field 2: map of key/value
-     */
-    @SuppressWarnings("rawtypes")
-    @Override
-    public List<Integer> emit(List<Object> tuple, Object messageId) {
-        if (!sanityCheck()) {
-            LOG.error(
-                    "spout collector for topic {} see monitored metadata invalid, is this data source removed! Trigger message id {} ",
-                    topic, messageId);
-            return null;
-        }
-
-        KafkaMessageIdWrapper newMessageId = new KafkaMessageIdWrapper(messageId);
-        newMessageId.topic = topic;
-        /**
-            phase 1: tuple to stream converter
-            if this topic multiplexes multiple streams, then retrieve the individual streams
-        */
-        List<Object> convertedTuple = converter.convert(tuple);
-        if(convertedTuple == null) {
-            LOG.warn("source data {} can't be converted to a stream, ignore this message", tuple);
-            spout.ack(newMessageId);
-            return null;
-        }
-        Map m = (Map)convertedTuple.get(3);
-        Object streamId = convertedTuple.get(1);
-
-        StreamDefinition sd = sds.get(streamId);
-        if(sd == null){
-            LOG.warn("StreamDefinition {} is not found within {}, ignore this message", streamId, sds);
-            spout.ack(newMessageId);
-            return null;
-        }
-
-        StreamEvent event = convertToStreamEventByStreamDefinition((Long)convertedTuple.get(2), m, sds.get(streamId));
-        /*
-            phase 2: stream repartition
-        */
-        for(StreamRepartitionMetadata md : streamRepartitionMetadataList) {
-            // one stream may have multiple group-by strategies, each strategy is for a specific group-by
-            for(StreamRepartitionStrategy groupingStrategy : md.groupingStrategies){
-                int hash = 0;
-                if(groupingStrategy.getPartition().getType().equals(StreamPartition.Type.GROUPBY)) {
-                    hash = getRoutingHashByGroupingStrategy(m, groupingStrategy);
-                }else if(groupingStrategy.getPartition().getType().equals(StreamPartition.Type.SHUFFLE)){
-                    hash = Math.abs((int)System.currentTimeMillis());
-                }
-                int mod = hash % groupingStrategy.numTotalParticipatingRouterBolts;
-                // filter out message
-                if (mod >= groupingStrategy.startSequence && mod < groupingStrategy.startSequence + numOfRouterBolts) {
-                    // framework takes care of field grouping instead of using storm internal field grouping
-                    String sid = StreamIdConversion.generateStreamIdBetween(spout.getSpoutName(), spout.getRouteBoltName()+ (hash % numOfRouterBolts));
-                    if (LOG.isDebugEnabled()) {
-                        LOG.debug("Emitted tuple: {} with message Id: {}, with topic {}, to streamId {}", convertedTuple, messageId, topic, sid);
-                    }
-                    // send message to StreamRouterBolt
-                    PartitionedEvent pEvent = new PartitionedEvent(event, groupingStrategy.partition, hash);
-                    if(this.serializer == null){
-                         delegate.emit(sid, Collections.singletonList(pEvent), newMessageId);
-                    }else {
-                        try {
-                            delegate.emit(sid, Collections.singletonList(serializer.serialize(pEvent)), newMessageId);
-                        } catch (IOException e) {
-                            LOG.error("Failed to serialize {}", pEvent, e);
-                            throw new RuntimeException(e);
-                        }
-                    }
-                }else{
-                    // ******* short-cut ack ********
-                    // we should simply ack those messages which are not processed in this topology because KafkaSpout implementation requires _pending is empty
-                    // before moving to next offsets.
-                    if(LOG.isDebugEnabled()){
-                        LOG.debug("Message filtered with mod {} not within range {} and {} for message {}", mod, groupingStrategy.startSequence,
-                                groupingStrategy.startSequence+ numOfRouterBolts, tuple);
-                    }
-                    spout.ack(newMessageId);
-                }
-            }
-        }
-
-        return null;
-    }
-
-    @SuppressWarnings("rawtypes")
-    private int getRoutingHashByGroupingStrategy(Map data, StreamRepartitionStrategy gs){
-        // calculate hash value for values from group-by fields
-        HashCodeBuilder hashCodeBuilder = new HashCodeBuilder();
-        for(String groupingField : gs.partition.getColumns()) {
-            if(data.get(groupingField) != null){
-                hashCodeBuilder.append(data.get(groupingField));
-            } else {
-                LOG.warn("Required GroupBy fields {} not found: {}", gs.partition.getColumns(), data);
-            }
-        }
-        int hash = hashCodeBuilder.toHashCode();
-        hash = Math.abs(hash);
-        return hash;
-    }
-
-    private boolean sanityCheck() {
-        boolean isOk = true;
-        if (streamRepartitionMetadataList == null) {
-            LOG.error("streamRepartitionMetadataList is null!");
-            isOk = false;
-        }
-        if (converter == null) {
-            LOG.error("tuple2StreamMetadata is null!");
-            isOk = false;
-        }
-        return isOk;
-    }
-
-    @SuppressWarnings({ "rawtypes", "unchecked" })
-    private StreamEvent convertToStreamEventByStreamDefinition(long timestamp, Map m, StreamDefinition sd){
-        return StreamEvent.Builder().timestamep(timestamp).attributes(m,sd).build();
-    }
-
-    /**
-     * SpoutSpec may be changed, this class will respond to changes on tuple2StreamMetadataMap and streamRepartitionMetadataMap
-     * @param spoutSpec
-     * @param sds
-     */
-    @Override
-    public void update(SpoutSpec spoutSpec, Map<String, StreamDefinition> sds) {
-        this.streamRepartitionMetadataList = spoutSpec.getStreamRepartitionMetadataMap().get(topic);
-        this.converter = new Tuple2StreamConverter(spoutSpec.getTuple2StreamMetadataMap().get(topic));
-        this.sds = sds;
-    }
-
-    @Override
-    public StreamDefinition getStreamDefinition(String streamId) {
-        return this.sds.get(streamId);
-    }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/utils/CompressionUtils.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/utils/CompressionUtils.java b/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/utils/CompressionUtils.java
deleted file mode 100644
index f526cad..0000000
--- a/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/utils/CompressionUtils.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.utils;
-
-import com.google.common.io.ByteStreams;
-
-import java.io.*;
-import java.util.zip.GZIPInputStream;
-import java.util.zip.GZIPOutputStream;
-
-
-public class CompressionUtils {
-    public static byte[] compress(byte[] source) throws IOException {
-        if (source == null || source.length == 0) {
-            return source;
-        }
-        ByteArrayInputStream sourceStream = new ByteArrayInputStream(source);
-        ByteArrayOutputStream outputStream = new ByteArrayOutputStream(source.length / 2);
-        try (OutputStream compressor = new GZIPOutputStream(outputStream)) {
-            ByteStreams.copy(sourceStream, compressor);
-            compressor.close();
-        }
-        try {
-            return outputStream.toByteArray();
-        } finally {
-            sourceStream.close();
-            outputStream.close();
-        }
-    }
-
-    public static byte[] decompress(byte[] compressed) throws IOException{
-        if (compressed == null || compressed.length == 0) {
-            return compressed;
-        }
-        ByteArrayInputStream sourceStream = new ByteArrayInputStream(compressed);
-        ByteArrayOutputStream outputStream = new ByteArrayOutputStream(compressed.length * 2);
-        try (GZIPInputStream compressor = new GZIPInputStream(sourceStream)) {
-            ByteStreams.copy(compressor, outputStream);
-            compressor.close();
-        }
-        try {
-            return outputStream.toByteArray();
-        } finally {
-            sourceStream.close();
-            outputStream.close();
-        }
-    }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/utils/MetadataSerDeser.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/utils/MetadataSerDeser.java b/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/utils/MetadataSerDeser.java
deleted file mode 100644
index a576404..0000000
--- a/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/utils/MetadataSerDeser.java
+++ /dev/null
@@ -1,93 +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.utils;
-
-import java.io.InputStream;
-
-import org.codehaus.jackson.JsonParser;
-import org.codehaus.jackson.map.ObjectMapper;
-import org.codehaus.jackson.type.TypeReference;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Since 5/6/16.
- */
-public class MetadataSerDeser {
-    private static final Logger LOG = LoggerFactory.getLogger(MetadataSerDeser.class);
-
-    @SuppressWarnings("rawtypes")
-    public static <K> K deserialize(InputStream is, TypeReference typeRef){
-        ObjectMapper mapper = new ObjectMapper();
-        try {
-            K spec = mapper.readValue(is, typeRef);
-            return spec;
-        }catch(Exception ex){
-            LOG.error("error in deserializing metadata of type {} from input stream", new TypeReference<K>(){}.getType().getTypeName(), ex);
-        }
-        return null;
-    }
-
-    public static <K> K deserialize(InputStream is, Class<K> cls){
-        ObjectMapper mapper = new ObjectMapper();
-        mapper.configure(JsonParser.Feature.ALLOW_COMMENTS,true);
-        try {
-            K spec = mapper.readValue(is, cls);
-            return spec;
-        }catch(Exception ex){
-            LOG.error("Got error to deserialize metadata of type {} from input stream", new TypeReference<K>(){}.getType().getTypeName(), ex);
-        }
-        return null;
-    }
-
-    @SuppressWarnings("rawtypes")
-    public static <K> K deserialize(String json, TypeReference typeRef){
-        ObjectMapper mapper = new ObjectMapper();
-        try {
-            K spec = mapper.readValue(json, typeRef);
-            return spec;
-        }catch(Exception ex){
-            LOG.error("error in deserializing metadata of type {} from {}", new TypeReference<K>(){}.getType().getTypeName(), json, ex);
-        }
-        return null;
-    }
-
-    public static <K> K deserialize(String json, Class<K> cls){
-        ObjectMapper mapper = new ObjectMapper();
-        try {
-            K spec = mapper.readValue(json, cls);
-            return spec;
-        }catch(Exception ex){
-            LOG.error("error in deserializing metadata of type {} from {}", new TypeReference<K>(){}.getType().getTypeName(), json, ex);
-        }
-        return null;
-    }
-
-    public static <K> String serialize(K spec){
-        ObjectMapper mapper = new ObjectMapper();
-        try{
-            String json = mapper.writeValueAsString(spec);
-            return json;
-        }catch(Exception ex){
-            LOG.error("error in serializing object {} with type {}", spec, new TypeReference<K>(){}.getType().getTypeName(), ex);
-        }
-        return null;
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/737e5a2d/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/utils/SerializableUtils.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/utils/SerializableUtils.java b/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/utils/SerializableUtils.java
deleted file mode 100644
index f4652a3..0000000
--- a/eagle-core/eagle-alert/alert/alert-engine/alert-engine-base/src/main/java/org/apache/eagle/alert/engine/utils/SerializableUtils.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.utils;
-
-import java.io.ByteArrayInputStream;
-import java.io.ByteArrayOutputStream;
-import java.io.IOException;
-import java.io.ObjectInputStream;
-import java.io.ObjectOutputStream;
-import java.io.Serializable;
-
-import org.xerial.snappy.SnappyInputStream;
-import org.xerial.snappy.SnappyOutputStream;
-
-/**
- * Utilities for working with Serializables.
- *
- * Derived from "com.google.cloud.dataflow.sdk.util.SerializableUtils":
- * https://github.com/apache/incubator-beam/blob/master/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/SerializableUtils.java
- */
-public class SerializableUtils {
-  /**
-   * Serializes the argument into an array of bytes, and returns it.
-   *
-   * @throws IllegalArgumentException if there are errors when serializing
-   */
-  public static byte[] serializeToCompressedByteArray(Object value) {
-    try {
-      ByteArrayOutputStream buffer = new ByteArrayOutputStream();
-      try (ObjectOutputStream oos = new ObjectOutputStream(new SnappyOutputStream(buffer))) {
-        oos.writeObject(value);
-      }
-      return buffer.toByteArray();
-    } catch (IOException exn) {
-      throw new IllegalArgumentException(
-          "unable to serialize " + value,
-          exn);
-    }
-  }
-
-  /**
-   * Serializes the argument into an array of bytes, and returns it.
-   *
-   * @throws IllegalArgumentException if there are errors when serializing
-   */
-  public static byte[] serializeToByteArray(Object value) {
-    try {
-      ByteArrayOutputStream buffer = new ByteArrayOutputStream();
-      try (ObjectOutputStream oos = new ObjectOutputStream(buffer)) {
-        oos.writeObject(value);
-      }
-      return buffer.toByteArray();
-    } catch (IOException exn) {
-      throw new IllegalArgumentException("unable to serialize " + value, exn);
-    }
-  }
-
-  /**
-   * Deserializes an object from the given array of bytes, e.g., as
-   * serialized using {@link #serializeToCompressedByteArray}, and returns it.
-   *
-   * @throws IllegalArgumentException if there are errors when
-   * deserializing, using the provided description to identify what
-   * was being deserialized
-   */
-  public static Object deserializeFromByteArray(byte[] encodedValue,
-                                                          String description) {
-    try {
-      try (ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(encodedValue))) {
-        return ois.readObject();
-      }
-    } catch (IOException | ClassNotFoundException exn) {
-      throw new IllegalArgumentException(
-          "unable to deserialize " + description,
-          exn);
-    }
-  }
-
-  /**
-   * Deserializes an object from the given array of bytes, e.g., as
-   * serialized using {@link #serializeToCompressedByteArray}, and returns it.
-   *
-   * @throws IllegalArgumentException if there are errors when
-   * deserializing, using the provided description to identify what
-   * was being deserialized
-   */
-  public static Object deserializeFromCompressedByteArray(byte[] encodedValue,
-                                                          String description) {
-    try {
-      try (ObjectInputStream ois = new ObjectInputStream(
-          new SnappyInputStream(new ByteArrayInputStream(encodedValue)))) {
-        return ois.readObject();
-      }
-    } catch (IOException | ClassNotFoundException exn) {
-      throw new IllegalArgumentException(
-          "unable to deserialize " + description,
-          exn);
-    }
-  }
-
-  public static <T extends Serializable> T ensureSerializable(T value) {
-    @SuppressWarnings("unchecked")
-    T copy = (T) deserializeFromCompressedByteArray(serializeToCompressedByteArray(value),
-        value.toString());
-    return copy;
-  }
-
-  public static <T extends Serializable> T clone(T value) {
-    @SuppressWarnings("unchecked")
-    T copy = (T) deserializeFromCompressedByteArray(serializeToCompressedByteArray(value),
-        value.toString());
-    return copy;
-  }
-}
\ No newline at end of file



Mime
View raw message