eagle-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From h..@apache.org
Subject [20/84] [partial] eagle git commit: Clean repo for eagle site
Date Mon, 03 Apr 2017 11:54:28 GMT
http://git-wip-us.apache.org/repos/asf/eagle/blob/6fd95d5c/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/sorter/StreamWindowRepository.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/sorter/StreamWindowRepository.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/sorter/StreamWindowRepository.java
deleted file mode 100755
index 7a503e1..0000000
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/sorter/StreamWindowRepository.java
+++ /dev/null
@@ -1,252 +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;
-
-import org.apache.eagle.alert.engine.sorter.impl.StreamSortedWindowInMapDB;
-import org.apache.eagle.alert.engine.sorter.impl.StreamSortedWindowOnHeap;
-import com.google.common.base.Preconditions;
-import org.mapdb.DB;
-import org.mapdb.DBMaker;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.File;
-import java.io.IOException;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.UUID;
-
-/**
- * ===== Benchmark Result Report =====<br/><br/>
- *
- * <p>Num. Operation   Type                            Time<br/>
- * ---- ---------   ----                            ----<br/>
- * 1000    FlushTime    DIRECT_MEMORY                :    55<br/>
- * 1000    FlushTime    FILE_RAF                     :    63<br/>
- * 1000    FlushTime    MEMORY                       :    146<br/>
- * 1000    FlushTime    ONHEAP                       :    17<br/>
- * 1000    InsertTime    DIRECT_MEMORY               :    68<br/>
- * 1000    InsertTime    FILE_RAF                    :    223<br/>
- * 1000    InsertTime    MEMORY                      :    273<br/>
- * 1000    InsertTime    ONHEAP                      :    20<br/>
- * 10000    FlushTime    DIRECT_MEMORY               :    551<br/>
- * 10000    FlushTime    FILE_RAF                    :    668<br/>
- * 10000    FlushTime    MEMORY                      :    643<br/>
- * 10000    FlushTime    ONHEAP                      :    5<br/>
- * 10000    InsertTime    DIRECT_MEMORY              :    446<br/>
- * 10000    InsertTime    FILE_RAF                   :    2095<br/>
- * 10000    InsertTime    MEMORY                     :    784<br/>
- * 10000    InsertTime    ONHEAP                     :    29<br/>
- * 100000    FlushTime    DIRECT_MEMORY              :    6139<br/>
- * 100000    FlushTime    FILE_RAF                   :    6237<br/>
- * 100000    FlushTime    MEMORY                     :    6238<br/>
- * 100000    FlushTime    ONHEAP                     :    18<br/>
- * 100000    InsertTime    DIRECT_MEMORY             :    4499<br/>
- * 100000    InsertTime    FILE_RAF                  :    22343<br/>
- * 100000    InsertTime    MEMORY                    :    4962<br/>
- * 100000    InsertTime    ONHEAP                    :    107<br/>
- * 1000000    FlushTime    DIRECT_MEMORY             :    61356<br/>
- * 1000000    FlushTime    FILE_RAF                  :    63025<br/>
- * 1000000    FlushTime    MEMORY                    :    61380<br/>
- * 1000000    FlushTime    ONHEAP                    :    47<br/>
- * 1000000    InsertTime    DIRECT_MEMORY            :    43637<br/>
- * 1000000    InsertTime    FILE_RAF                 :    464481<br/>
- * 1000000    InsertTime    MEMORY                   :    44367<br/>
- * 1000000    InsertTime    ONHEAP                   :    2040<br/>
- * </p>
- * @see StreamSortedWindowOnHeap
- * @see org.mapdb.DBMaker
- */
-public class StreamWindowRepository {
-    public enum StorageType {
-        /**
-         * Creates new in-memory database which stores all data on heap without serialization.
-         * This mode should be very fast, but data will affect Garbage PartitionedEventCollector the same way as traditional Java Collections.
-         */
-        ONHEAP,
-
-        /**
-         * Creates new in-memory database. Changes are lost after JVM exits.
-         * This option serializes data into {@code byte[]},
-         * so they are not affected by Garbage PartitionedEventCollector.
-         */
-        MEMORY,
-
-        /**
-         * <p>
-         * Creates new in-memory database. Changes are lost after JVM exits.
-         * </p><p>
-         * This will use {@code DirectByteBuffer} outside of HEAP, so Garbage Collector is not affected
-         * You should increase ammount of direct memory with
-         * {@code -XX:MaxDirectMemorySize=10G} JVM param
-         * </p>
-         */
-        DIRECT_MEMORY,
-
-        /**
-         * By default use File.createTempFile("streamwindows","temp")
-         */
-        FILE_RAF
-    }
-
-    private static final Logger LOG = LoggerFactory.getLogger(StreamWindowRepository.class);
-    private final Map<StorageType, DB> dbPool;
-
-    private StreamWindowRepository() {
-        dbPool = new HashMap<>();
-    }
-
-    private static StreamWindowRepository repository;
-
-    /**
-     * Close automatically when JVM exists.
-     *
-     * @return StreamWindowRepository singletonInstance
-     */
-    public static StreamWindowRepository getSingletonInstance() {
-        synchronized (StreamWindowRepository.class) {
-            if (repository == null) {
-                repository = new StreamWindowRepository();
-                Runtime.getRuntime().addShutdownHook(new Thread() {
-                    @Override
-                    public void run() {
-                        repository.close();
-                    }
-                });
-            }
-            return repository;
-        }
-    }
-
-    private DB createMapDB(StorageType storageType) {
-        synchronized (dbPool) {
-            if (!dbPool.containsKey(storageType)) {
-                DB db;
-                switch (storageType) {
-                    case ONHEAP:
-                        db = DBMaker.heapDB().closeOnJvmShutdown().make();
-                        LOG.info("Create ONHEAP mapdb");
-                        break;
-                    case MEMORY:
-                        db = DBMaker.memoryDB().closeOnJvmShutdown().make();
-                        LOG.info("Create MEMORY mapdb");
-                        break;
-                    case DIRECT_MEMORY:
-                        db = DBMaker.memoryDirectDB().closeOnJvmShutdown().make();
-                        LOG.info("Create DIRECT_MEMORY mapdb");
-                        break;
-                    case FILE_RAF:
-                        try {
-                            File file = File.createTempFile("window-", ".map");
-                            file.delete();
-                            file.deleteOnExit();
-                            Preconditions.checkNotNull(file, "file is null");
-                            db = DBMaker.fileDB(file).deleteFilesAfterClose().make();
-                            LOG.info("Created FILE_RAF map file at {}", file.getAbsolutePath());
-                        } catch (IOException e) {
-                            throw new IllegalStateException(e);
-                        }
-                        break;
-                    default:
-                        throw new IllegalArgumentException("Illegal storage type: " + storageType);
-                }
-                dbPool.put(storageType, db);
-                return db;
-            }
-            return dbPool.get(storageType);
-        }
-    }
-
-    public StreamWindow createWindow(long start, long end, long margin, StorageType type) {
-        StreamWindow ret;
-        switch (type) {
-            case ONHEAP:
-                ret = new StreamSortedWindowOnHeap(start, end, margin);
-                break;
-            default:
-                ret = new StreamSortedWindowInMapDB(
-                    start, end, margin,
-                    createMapDB(type),
-                    UUID.randomUUID().toString()
-                );
-                break;
-        }
-
-        if (LOG.isDebugEnabled()) {
-            LOG.debug("Created new {}, type: {}", ret, type);
-        }
-        return ret;
-    }
-
-    public StreamWindow createWindow(long start, long end, long margin, StreamWindowStrategy strategy) {
-        return strategy.createWindow(start, end, margin, this);
-    }
-
-    public StreamWindow createWindow(long start, long end, long margin) {
-        return OnHeapStrategy.INSTANCE.createWindow(start, end, margin, this);
-    }
-
-    public void close() {
-        for (Map.Entry<StorageType, DB> entry : dbPool.entrySet()) {
-            entry.getValue().close();
-        }
-        dbPool.clear();
-    }
-
-    public interface StreamWindowStrategy {
-        StreamWindow createWindow(long start, long end, long margin, StreamWindowRepository repository);
-    }
-
-    public static class OnHeapStrategy implements StreamWindowStrategy {
-        public static final OnHeapStrategy INSTANCE = new OnHeapStrategy();
-
-        @Override
-        public StreamWindow createWindow(long start, long end, long margin, StreamWindowRepository repository) {
-            return repository.createWindow(start, end, margin, StorageType.ONHEAP);
-        }
-    }
-
-    public static class WindowSizeStrategy implements StreamWindowStrategy {
-        private static final long ONE_HOUR = 3600 * 1000;
-        private static final long FIVE_HOURS = 5 * 3600 * 1000;
-        private final long onheapWindowSizeLimit;
-        private final long offheapWindowSizeLimit;
-
-        public static WindowSizeStrategy INSTANCE = new WindowSizeStrategy(ONE_HOUR, FIVE_HOURS);
-
-        public WindowSizeStrategy(long onheapWindowSizeLimit, long offheapWindowSizeLimit) {
-            this.offheapWindowSizeLimit = offheapWindowSizeLimit;
-            this.onheapWindowSizeLimit = onheapWindowSizeLimit;
-
-            if (this.offheapWindowSizeLimit < this.onheapWindowSizeLimit) {
-                throw new IllegalStateException("offheapWindowSizeLimit " + this.offheapWindowSizeLimit + " < onheapWindowSizeLimit " + this.onheapWindowSizeLimit);
-            }
-        }
-
-        @Override
-        public StreamWindow createWindow(long start, long end, long margin, StreamWindowRepository repository) {
-            long windowLength = end - start;
-            if (windowLength <= onheapWindowSizeLimit) {
-                return repository.createWindow(start, end, margin, StreamWindowRepository.StorageType.ONHEAP);
-            } else if (windowLength > onheapWindowSizeLimit & windowLength <= offheapWindowSizeLimit) {
-                return repository.createWindow(start, end, margin, StreamWindowRepository.StorageType.DIRECT_MEMORY);
-            } else {
-                return repository.createWindow(start, end, margin, StreamWindowRepository.StorageType.FILE_RAF);
-            }
-        }
-    }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/eagle/blob/6fd95d5c/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/sorter/impl/CachedEventGroupSerializer.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/sorter/impl/CachedEventGroupSerializer.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/sorter/impl/CachedEventGroupSerializer.java
deleted file mode 100644
index 73adee6..0000000
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/sorter/impl/CachedEventGroupSerializer.java
+++ /dev/null
@@ -1,115 +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 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.utils.SerializableUtils;
-import org.apache.commons.lang.builder.HashCodeBuilder;
-import org.jetbrains.annotations.NotNull;
-import org.mapdb.DataInput2;
-import org.mapdb.DataOutput2;
-import org.mapdb.serializer.GroupSerializerObjectArray;
-
-import java.io.IOException;
-import java.util.HashMap;
-import java.util.Map;
-
-/**
- * @deprecated performance is worse, should investigate.
- */
-public class CachedEventGroupSerializer extends GroupSerializerObjectArray<PartitionedEvent[]> {
-    private Map<Integer, StreamPartition> hashCodePartitionDict = new HashMap<>();
-
-    private void writePartitionedEvent(DataOutput2 out, PartitionedEvent event) throws IOException {
-        out.packLong(event.getPartitionKey());
-        int partitionHashCode = 0;
-        if (event.getPartition() != null) {
-            partitionHashCode = event.getPartition().hashCode();
-            if (!hashCodePartitionDict.containsKey(partitionHashCode)) {
-                hashCodePartitionDict.put(partitionHashCode, event.getPartition());
-            }
-        }
-        out.packInt(partitionHashCode);
-        if (event.getEvent() != null) {
-            byte[] eventBytes = SerializableUtils.serializeToCompressedByteArray(event.getEvent());
-            out.packInt(eventBytes.length);
-            out.write(eventBytes);
-        } else {
-            out.packInt(0);
-        }
-    }
-
-    private PartitionedEvent readPartitionedEvent(DataInput2 in) throws IOException {
-        PartitionedEvent event = new PartitionedEvent();
-        event.setPartitionKey(in.unpackLong());
-        int partitionHashCode = in.unpackInt();
-        if (partitionHashCode != 0 && hashCodePartitionDict.containsKey(partitionHashCode)) {
-            event.setPartition(hashCodePartitionDict.get(partitionHashCode));
-        }
-        int eventBytesLen = in.unpackInt();
-        if (eventBytesLen > 0) {
-            byte[] eventBytes = new byte[eventBytesLen];
-            in.readFully(eventBytes);
-            event.setEvent((StreamEvent) SerializableUtils.deserializeFromCompressedByteArray(eventBytes, "Deserialize event from bytes"));
-        }
-        return event;
-    }
-
-    @Override
-    public void serialize(DataOutput2 out, PartitionedEvent[] value) throws IOException {
-        out.packInt(value.length);
-        for (PartitionedEvent event : value) {
-            writePartitionedEvent(out, event);
-        }
-    }
-
-    @Override
-    public PartitionedEvent[] deserialize(DataInput2 in, int available) throws IOException {
-        final int size = in.unpackInt();
-        PartitionedEvent[] ret = new PartitionedEvent[size];
-        for (int i = 0; i < size; i++) {
-            ret[i] = readPartitionedEvent(in);
-        }
-        return ret;
-    }
-
-    @Override
-    public boolean isTrusted() {
-        return true;
-    }
-
-    @Override
-    public boolean equals(PartitionedEvent[] a1, PartitionedEvent[] a2) {
-        return a1[0].getTimestamp() == a2[0].getTimestamp();
-    }
-
-    @Override
-    public int hashCode(@NotNull PartitionedEvent[] events, int seed) {
-        return new HashCodeBuilder().append(events).toHashCode();
-    }
-
-    @Override
-    public int compare(PartitionedEvent[] o1, PartitionedEvent[] o2) {
-        if (o1.length > 0 && o2.length > 0) {
-            return (int) (o1[0].getTimestamp() - o2[0].getTimestamp());
-        } else {
-            return 0;
-        }
-    }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/eagle/blob/6fd95d5c/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/sorter/impl/PartitionedEventGroupSerializer.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/sorter/impl/PartitionedEventGroupSerializer.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/sorter/impl/PartitionedEventGroupSerializer.java
deleted file mode 100644
index 55efcaf..0000000
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/sorter/impl/PartitionedEventGroupSerializer.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 org.apache.eagle.alert.engine.model.PartitionedEvent;
-import org.apache.eagle.alert.engine.utils.SerializableUtils;
-import org.jetbrains.annotations.NotNull;
-import org.mapdb.DataInput2;
-import org.mapdb.DataOutput2;
-import org.mapdb.Serializer;
-import org.mapdb.serializer.GroupSerializer;
-
-import java.io.IOException;
-import java.util.Comparator;
-
-
-public class PartitionedEventGroupSerializer implements GroupSerializer<PartitionedEvent[]> {
-    private static final GroupSerializer<byte[]> delegate = Serializer.BYTE_ARRAY;
-
-    @Override
-    public int valueArraySearch(Object keys, PartitionedEvent[] key) {
-        return delegate.valueArraySearch(keys, serialize(key));
-    }
-
-    @SuppressWarnings("rawtypes")
-    @Override
-    public int valueArraySearch(Object keys, PartitionedEvent[] key, Comparator comparator) {
-        return delegate.valueArraySearch(keys, serialize(key), comparator);
-    }
-
-    @Override
-    public void valueArraySerialize(DataOutput2 out, Object vals) throws IOException {
-        delegate.valueArraySerialize(out, vals);
-    }
-
-    @Override
-    public Object valueArrayDeserialize(DataInput2 in, int size) throws IOException {
-        return delegate.valueArrayDeserialize(in, size);
-    }
-
-    @Override
-    public PartitionedEvent[] valueArrayGet(Object vals, int pos) {
-        return deserialize(delegate.valueArrayGet(vals, pos));
-    }
-
-    @Override
-    public int valueArraySize(Object vals) {
-        return delegate.valueArraySize(vals);
-    }
-
-    @Override
-    public Object valueArrayEmpty() {
-        return delegate.valueArrayEmpty();
-    }
-
-    @Override
-    public Object valueArrayPut(Object vals, int pos, PartitionedEvent[] newValue) {
-        return delegate.valueArrayPut(vals, pos, serialize(newValue));
-    }
-
-    @Override
-    public Object valueArrayUpdateVal(Object vals, int pos, PartitionedEvent[] newValue) {
-        return delegate.valueArrayUpdateVal(vals, pos, serialize(newValue));
-    }
-
-    @Override
-    public Object valueArrayFromArray(Object[] objects) {
-        return delegate.valueArrayFromArray(objects);
-    }
-
-    @Override
-    public Object valueArrayCopyOfRange(Object vals, int from, int to) {
-        return delegate.valueArrayCopyOfRange(vals, from, to);
-    }
-
-    @Override
-    public Object valueArrayDeleteValue(Object vals, int pos) {
-        return delegate.valueArrayDeleteValue(vals, pos);
-    }
-
-    @Override
-    public void serialize(@NotNull DataOutput2 out, @NotNull PartitionedEvent[] value) throws IOException {
-        delegate.serialize(out, serialize(value));
-    }
-
-    private static byte[] serialize(PartitionedEvent[] events) {
-        return SerializableUtils.serializeToCompressedByteArray(events);
-    }
-
-    @Override
-    public PartitionedEvent[] deserialize(@NotNull DataInput2 input, int available) throws IOException {
-        return deserialize(delegate.deserialize(input, available));
-    }
-
-    private static PartitionedEvent[] deserialize(byte[] bytes) {
-        return (PartitionedEvent[]) SerializableUtils.deserializeFromCompressedByteArray(bytes, "deserialize as stream event");
-    }
-
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/eagle/blob/6fd95d5c/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/sorter/impl/PartitionedEventTimeOrderingComparator.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/sorter/impl/PartitionedEventTimeOrderingComparator.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/sorter/impl/PartitionedEventTimeOrderingComparator.java
deleted file mode 100644
index 5378c67..0000000
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/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 org.apache.eagle.alert.engine.model.PartitionedEvent;
-
-import java.util.Comparator;
-import java.util.Objects;
-
-/**
- * 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/eagle/blob/6fd95d5c/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/sorter/impl/StreamSortWindowHandlerImpl.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/sorter/impl/StreamSortWindowHandlerImpl.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/sorter/impl/StreamSortWindowHandlerImpl.java
deleted file mode 100644
index fb5ba72..0000000
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/sorter/impl/StreamSortWindowHandlerImpl.java
+++ /dev/null
@@ -1,114 +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 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.common.DateTimeUtil;
-
-import org.joda.time.Period;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-
-public class StreamSortWindowHandlerImpl implements StreamSortHandler {
-    private static final 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/eagle/blob/6fd95d5c/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/sorter/impl/StreamSortedWindowInMapDB.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/sorter/impl/StreamSortedWindowInMapDB.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/sorter/impl/StreamSortedWindowInMapDB.java
deleted file mode 100644
index 73a63b4..0000000
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/sorter/impl/StreamSortedWindowInMapDB.java
+++ /dev/null
@@ -1,139 +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 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.apache.commons.lang3.time.StopWatch;
-import org.mapdb.BTreeMap;
-import org.mapdb.DB;
-import org.mapdb.Serializer;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.Arrays;
-import java.util.concurrent.atomic.AtomicInteger;
-
-/**
- * StreamSortedWindow based on MapDB to support off-heap or disk storage.
- * Stable sorting algorithm
- * 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 static final Logger LOG = LoggerFactory.getLogger(StreamSortedWindowInMapDB.class);
-    private final AtomicInteger size;
-    private long replaceOpCount = 0;
-    private static final PartitionedEventGroupSerializer STREAM_EVENT_GROUP_SERIALIZER = new PartitionedEventGroupSerializer();
-
-    /**
-     * @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/eagle/blob/6fd95d5c/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/sorter/impl/StreamSortedWindowOnHeap.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/sorter/impl/StreamSortedWindowOnHeap.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/sorter/impl/StreamSortedWindowOnHeap.java
deleted file mode 100644
index ed000f1..0000000
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/sorter/impl/StreamSortedWindowOnHeap.java
+++ /dev/null
@@ -1,82 +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 org.apache.eagle.alert.engine.PartitionedEventCollector;
-import org.apache.eagle.alert.engine.model.PartitionedEvent;
-import org.apache.eagle.alert.engine.sorter.BaseStreamWindow;
-
-import com.google.common.collect.TreeMultiset;
-import org.apache.commons.lang3.time.StopWatch;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.Comparator;
-
-public class StreamSortedWindowOnHeap extends BaseStreamWindow {
-    private static final 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/eagle/blob/6fd95d5c/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/sorter/impl/StreamTimeClockInLocalMemory.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/sorter/impl/StreamTimeClockInLocalMemory.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/sorter/impl/StreamTimeClockInLocalMemory.java
deleted file mode 100644
index e5be786..0000000
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/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 org.apache.eagle.alert.engine.sorter.StreamTimeClock;
-import org.apache.eagle.common.DateTimeUtil;
-
-import java.util.concurrent.atomic.AtomicLong;
-
-
-/**
- * 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/eagle/blob/6fd95d5c/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/sorter/impl/StreamTimeClockManagerImpl.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/sorter/impl/StreamTimeClockManagerImpl.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/sorter/impl/StreamTimeClockManagerImpl.java
deleted file mode 100644
index b59918d..0000000
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/sorter/impl/StreamTimeClockManagerImpl.java
+++ /dev/null
@@ -1,169 +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 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.common.DateTimeUtil;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Timer;
-import java.util.TimerTask;
-import java.util.concurrent.atomic.AtomicInteger;
-
-public final class StreamTimeClockManagerImpl implements StreamTimeClockManager {
-    private static final long serialVersionUID = -2770823821511195343L;
-    private static final Logger LOG = LoggerFactory.getLogger(StreamTimeClockManagerImpl.class);
-    private final Map<String, StreamTimeClock> streamIdTimeClockMap;
-    private Timer timer;
-
-    private final Map<StreamTimeClockListener, String> listenerStreamIdMap;
-    private static final 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);
-        }
-    }
-
-    @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);
-        }
-    }
-
-    @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/eagle/blob/6fd95d5c/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/sorter/impl/StreamWindowManagerImpl.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/sorter/impl/StreamWindowManagerImpl.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/sorter/impl/StreamWindowManagerImpl.java
deleted file mode 100644
index 4c5154b..0000000
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/sorter/impl/StreamWindowManagerImpl.java
+++ /dev/null
@@ -1,176 +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 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.TimePeriodUtils;
-import org.apache.eagle.common.DateTimeUtil;
-
-import org.apache.commons.lang3.time.StopWatch;
-import org.joda.time.Period;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.*;
-
-public class StreamWindowManagerImpl implements StreamWindowManager {
-    private static final 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/eagle/blob/6fd95d5c/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/spout/CorrelationSpout.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/spout/CorrelationSpout.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/spout/CorrelationSpout.java
deleted file mode 100644
index e9ee892..0000000
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/spout/CorrelationSpout.java
+++ /dev/null
@@ -1,411 +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.MultiScheme;
-import backtype.storm.spout.Scheme;
-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;
-import com.typesafe.config.ConfigFactory;
-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.*;
-
-import java.text.MessageFormat;
-import java.util.*;
-
-/**
- * wrap KafkaSpout to provide parallel processing of messages for multiple Kafka topics
- * <p>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</p>
- */
-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()) {
-            try {
-                wrapper.nextTuple();
-            } catch (Exception e) {
-                LOG.error("unexpected exception is caught: {}", e.getMessage(), e);
-            }
-
-        }
-    }
-
-    /**
-     * 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);
-        if (spout != null) {
-            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);
-        if (spout != null) {
-            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>();
-        Map<String, Map<String, String>> dataSourceProperties = new HashMap<>();
-        for (Kafka2TupleMetadata ds : newMeta.getKafka2TupleMetadataMap().values()) {
-            newSchemaName.put(ds.getTopic(), ds.getSchemeCls());
-            dataSourceProperties.put(ds.getTopic(), ds.getProperties());
-        }
-
-        // 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(ConfigFactory.parseMap(dataSourceProperties.get(topic)).withFallback(this.config),
-                    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
-     * <p>TODO: Should avoid use Config.get in deep calling stack, should generate config bean as early as possible
-     * </p>
-     *
-     * @param conf
-     * @param context
-     * @param collector
-     * @param topic
-     * @param spoutSpec
-     * @return
-     */
-    @SuppressWarnings("rawtypes")
-    protected KafkaSpoutWrapper createKafkaSpout(Config configure, Map conf, TopologyContext context, SpoutOutputCollector collector, final String topic,
-                                                 String schemeClsName, SpoutSpec spoutSpec, Map<String, StreamDefinition> sds) throws Exception {
-        String kafkaBrokerZkQuorum = configure.getString(AlertConstants.KAFKA_BROKER_ZK_QUORUM);
-        BrokerHosts hosts = null;
-        if (configure.hasPath("spout.kafkaBrokerZkBasePath")) {
-            hosts = new ZkHosts(kafkaBrokerZkQuorum, configure.getString(AlertConstants.KAFKA_BROKER_ZK_BASE_PATH));
-        } else {
-            hosts = new ZkHosts(kafkaBrokerZkQuorum);
-        }
-        String transactionZkRoot = DEFAULT_STORM_KAFKA_TRANSACTION_ZK_ROOT;
-        if (configure.hasPath("spout.stormKafkaTransactionZkPath")) {
-            transactionZkRoot = configure.getString("spout.stormKafkaTransactionZkPath");
-        }
-        boolean logEventEnabled = false;
-        if (configure.hasPath("topology.logEventEnabled")) {
-            logEventEnabled = configure.getBoolean("topology.logEventEnabled");
-        }
-        // write partition offset etc. into zkRoot+id, see PartitionManager.committedPath
-        String zkStateTransactionRelPath = DEFAULT_STORM_KAFKA_TRANSACTION_ZK_RELATIVE_PATH;
-        if (configure.hasPath("spout.stormKafkaEagleConsumer")) {
-            zkStateTransactionRelPath = configure.getString("spout.stormKafkaEagleConsumer");
-        }
-        SpoutConfig spoutConfig = new SpoutConfig(hosts, topic, transactionZkRoot, zkStateTransactionRelPath + "/" + topic + "/" + topologyId);
-        // transaction zkServers
-        boolean stormKafkaUseSameZkQuorumWithKafkaBroker = configure.getBoolean("spout.stormKafkaUseSameZkQuorumWithKafkaBroker");
-        if (stormKafkaUseSameZkQuorumWithKafkaBroker) {
-            ZkServerPortUtils utils = new ZkServerPortUtils(kafkaBrokerZkQuorum);
-            spoutConfig.zkServers = utils.getZkHosts();
-            spoutConfig.zkPort = utils.getZkPort();
-        } else {
-            ZkServerPortUtils utils = new ZkServerPortUtils(configure.getString("spout.stormKafkaTransactionZkQuorum"));
-            spoutConfig.zkServers = utils.getZkHosts();
-            spoutConfig.zkPort = utils.getZkPort();
-        }
-        // transaction update interval
-        spoutConfig.stateUpdateIntervalMs = configure.hasPath("spout.stormKafkaStateUpdateIntervalMs") ? configure.getInt("spout.stormKafkaStateUpdateIntervalMs") : 2000;
-        // Kafka fetch size
-        spoutConfig.fetchSizeBytes = configure.hasPath("spout.stormKafkaFetchSizeBytes") ? configure.getInt("spout.stormKafkaFetchSizeBytes") : 1048586;
-        // "startOffsetTime" is for test usage, prod should not use this
-        if (configure.hasPath("spout.stormKafkaStartOffsetTime")) {
-            spoutConfig.startOffsetTime = configure.getInt("spout.stormKafkaStartOffsetTime");
-        }
-
-        spoutConfig.scheme = createMultiScheme(conf, topic, schemeClsName);
-        KafkaSpoutWrapper wrapper = new KafkaSpoutWrapper(spoutConfig, kafkaSpoutMetric);
-        SpoutOutputCollectorWrapper collectorWrapper = new SpoutOutputCollectorWrapper(this, collector, topic, spoutSpec, numOfRouterBolts, sds, this.serializer, logEventEnabled);
-        wrapper.open(conf, context, collectorWrapper);
-
-        if (LOG.isInfoEnabled()) {
-            LOG.info("create and open kafka wrapper: topic {}, scheme class{} ", topic, schemeClsName);
-        }
-        return wrapper;
-    }
-
-    private MultiScheme createMultiScheme(Map conf, String topic, String schemeClsName) throws Exception {
-        Object scheme = SchemeBuilder.buildFromClsName(schemeClsName, topic, conf);
-        if (scheme instanceof MultiScheme) {
-            return (MultiScheme) scheme;
-        } else if (scheme instanceof Scheme) {
-            return new SchemeAsMultiScheme((Scheme) scheme);
-        } else {
-            LOG.error("create spout scheme failed.");
-            throw new IllegalArgumentException("create spout scheme failed.");
-        }
-    }
-
-    @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/eagle/blob/6fd95d5c/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/spout/CreateTopicUtils.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/spout/CreateTopicUtils.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/spout/CreateTopicUtils.java
deleted file mode 100644
index 5b7e542..0000000
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/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 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) {
-        //        ZkConnection zkConnection = new ZkConnection(zkQuorum);
-        //        ZkClient zkClient = new ZkClient(zkQuorum, 10000, 10000, ZKStringSerializer$.MODULE$);
-        ////        ZkUtils zkUtils = new ZkUtils(zkClient, zkConnection, false);
-        //        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/eagle/blob/6fd95d5c/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/spout/ISpoutSpecLCM.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/spout/ISpoutSpecLCM.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/spout/ISpoutSpecLCM.java
deleted file mode 100644
index 3c8c99d..0000000
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/spout/ISpoutSpecLCM.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 org.apache.eagle.alert.coordination.model.SpoutSpec;
-import org.apache.eagle.alert.engine.coordinator.StreamDefinition;
-
-import java.util.Map;
-
-/**
- * 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/eagle/blob/6fd95d5c/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/spout/KafkaMessageIdWrapper.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/spout/KafkaMessageIdWrapper.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/spout/KafkaMessageIdWrapper.java
deleted file mode 100644
index 74dea03..0000000
--- a/eagle-core/eagle-alert-parent/eagle-alert/alert-engine/src/main/java/org/apache/eagle/alert/engine/spout/KafkaMessageIdWrapper.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
- *  *
- *  *    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;
-import org.apache.eagle.common.DateTimeUtil;
-
-/**
- * Created on 2/18/16.
- */
-public class KafkaMessageIdWrapper {
-    public Object id;
-    public String topic;
-    public long timestamp;
-
-    public KafkaMessageIdWrapper(Object o) {
-        this.id = o;
-    }
-
-    private static final ObjectMapper objectMapper = new ObjectMapper();
-
-    public String toString() {
-        try {
-            return String.format("KafkaMessageIdWrapper[topic=%s, id=%s, timestamp=%s %s]",
-                    topic,
-                    objectMapper.writeValueAsString(id),
-                    DateTimeUtil.millisecondsToHumanDateWithSeconds(timestamp),
-                    DateTimeUtil.CURRENT_TIME_ZONE.getID());
-        } catch (JsonProcessingException e) {
-            throw new IllegalStateException(e);
-        }
-    }
-}
\ No newline at end of file

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


Mime
View raw message