diff --git a/Sources/Core/Microsoft.StreamProcessing/Aggregates/SortedMultisetAggregateBase.cs b/Sources/Core/Microsoft.StreamProcessing/Aggregates/SortedMultisetAggregateBase.cs
index 645f9f226..aa31c9dc0 100644
--- a/Sources/Core/Microsoft.StreamProcessing/Aggregates/SortedMultisetAggregateBase.cs
+++ b/Sources/Core/Microsoft.StreamProcessing/Aggregates/SortedMultisetAggregateBase.cs
@@ -17,7 +17,7 @@ protected SortedMultisetAggregateBase(IComparerExpression<T> comparer, QueryCont
             Expression<Func<Func<SortedDictionary<T, long>>, SortedMultiSet<T>>> template
                 = (g) => new SortedMultiSet<T>(g);
             var replaced = template.ReplaceParametersInBody(generator);
-            initialState = Expression.Lambda<Func<SortedMultiSet<T>>>(replaced);
+            this.initialState = Expression.Lambda<Func<SortedMultiSet<T>>>(replaced);
         }
 
         private readonly Expression<Func<SortedMultiSet<T>>> initialState;
diff --git a/Sources/Core/Microsoft.StreamProcessing/Aggregates/TopKAggregate.cs b/Sources/Core/Microsoft.StreamProcessing/Aggregates/TopKAggregate.cs
index 15ebdacff..2b07ee362 100644
--- a/Sources/Core/Microsoft.StreamProcessing/Aggregates/TopKAggregate.cs
+++ b/Sources/Core/Microsoft.StreamProcessing/Aggregates/TopKAggregate.cs
@@ -6,27 +6,37 @@
 using System.Collections.Generic;
 using System.Diagnostics.Contracts;
 using System.Linq.Expressions;
+using Microsoft.StreamProcessing.Internal;
 
 namespace Microsoft.StreamProcessing.Aggregates
 {
-    internal sealed class TopKAggregate<T> : SortedMultisetAggregateBase<T, List<RankedEvent<T>>>
+    internal sealed class TopKAggregate<T> : IAggregate<T, ITopKState<T>, List<RankedEvent<T>>>
     {
         private readonly Comparison<T> compiledRankComparer;
         private readonly int k;
 
-        public TopKAggregate(int k, QueryContainer container) : this(k, ComparerExpression<T>.Default, container) { }
+        public TopKAggregate(int k, IComparerExpression<T> rankComparer, QueryContainer container, long hoppingWindowSize)
+            : this(k, rankComparer, ComparerExpression<T>.Default, container, hoppingWindowSize) { }
 
-        public TopKAggregate(int k, IComparerExpression<T> rankComparer, QueryContainer container)
-            : this(k, rankComparer, ComparerExpression<T>.Default, container) { }
-
-        public TopKAggregate(int k, IComparerExpression<T> rankComparer, IComparerExpression<T> overallComparer, QueryContainer container)
-            : base(ThenOrderBy(Reverse(rankComparer), overallComparer), container)
+        public TopKAggregate(int k, IComparerExpression<T> rankComparer, IComparerExpression<T> overallComparer,
+            QueryContainer container, long hoppingWindowSize)
         {
             Contract.Requires(rankComparer != null);
             Contract.Requires(overallComparer != null);
             Contract.Requires(k > 0);
             this.compiledRankComparer = Reverse(rankComparer).GetCompareExpr().Compile();
             this.k = k;
+
+            Expression<Func<Func<SortedDictionary<T, long>>, ITopKState<T>>> template;
+            if (hoppingWindowSize > 0 && hoppingWindowSize < 1000000)
+                template = (g) => new HoppingTopKState<T>(k, compiledRankComparer, (int)hoppingWindowSize, g);
+            else
+                template = (g) => new SimpleTopKState<T>(g);
+
+            var combinedComparer = ThenOrderBy(Reverse(rankComparer), overallComparer);
+            var generator = combinedComparer.CreateSortedDictionaryGenerator<T, long>(container);
+            var replaced = template.ReplaceParametersInBody(generator);
+            this.initialState = Expression.Lambda<Func<ITopKState<T>>>(replaced);
         }
 
         private static IComparerExpression<T> Reverse(IComparerExpression<T> comparer)
@@ -53,10 +63,11 @@ private static IComparerExpression<T> ThenOrderBy(IComparerExpression<T> compare
             return new ComparerExpression<T>(newExpression);
         }
 
-        public override Expression<Func<SortedMultiSet<T>, List<RankedEvent<T>>>> ComputeResult() => set => GetTopK(set);
+        public Expression<Func<ITopKState<T>, List<RankedEvent<T>>>> ComputeResult() => set => GetTopK(set);
 
-        private List<RankedEvent<T>> GetTopK(SortedMultiSet<T> set)
+        private List<RankedEvent<T>> GetTopK(ITopKState<T> state)
         {
+            var set = state.GetSortedValues();
             int count = (int)Math.Min(this.k, set.TotalCount);
             var result = new List<RankedEvent<T>>(count);
             int nextRank = 1;
@@ -82,5 +93,20 @@ private List<RankedEvent<T>> GetTopK(SortedMultiSet<T> set)
 
             return result;
         }
+
+        private readonly Expression<Func<ITopKState<T>>> initialState;
+        public Expression<Func<ITopKState<T>>> InitialState() => initialState;
+
+        private static readonly Expression<Func<ITopKState<T>, long, T, ITopKState<T>>> acc
+            = (state, timestamp, input) => state.Add(input, timestamp);
+        public Expression<Func<ITopKState<T>, long, T, ITopKState<T>>> Accumulate() => acc;
+
+        private static readonly Expression<Func<ITopKState<T>, long, T, ITopKState<T>>> dec
+            = (state, timestamp, input) => state.Remove(input, timestamp);
+        public Expression<Func<ITopKState<T>, long, T, ITopKState<T>>> Deaccumulate() => dec;
+
+        private static readonly Expression<Func<ITopKState<T>, ITopKState<T>, ITopKState<T>>> diff
+            = (leftState, rightState) => leftState.RemoveAll(rightState);
+        public Expression<Func<ITopKState<T>, ITopKState<T>, ITopKState<T>>> Difference() => diff;
     }
 }
diff --git a/Sources/Core/Microsoft.StreamProcessing/Aggregates/TopKState.cs b/Sources/Core/Microsoft.StreamProcessing/Aggregates/TopKState.cs
new file mode 100644
index 000000000..4313df3a0
--- /dev/null
+++ b/Sources/Core/Microsoft.StreamProcessing/Aggregates/TopKState.cs
@@ -0,0 +1,230 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+
+namespace Microsoft.StreamProcessing.Aggregates
+{
+    /// <summary>
+    /// State used by TopK Aggregate
+    /// </summary>
+    /// <typeparam name="T"></typeparam>
+    public interface ITopKState<T>
+    {
+        /// <summary>
+        /// Add a single entry
+        /// </summary>
+        /// <param name="input"></param>
+        /// <param name="timestamp"></param>
+        ITopKState<T> Add(T input, long timestamp);
+
+        /// <summary>
+        /// Removes the specified entry
+        /// </summary>
+        /// <param name="input"></param>
+        /// <param name="timestamp"></param>
+        ITopKState<T> Remove(T input, long timestamp);
+
+        /// <summary>
+        /// Removes entries from other
+        /// </summary>
+        /// <param name="other"></param>
+        ITopKState<T> RemoveAll(ITopKState<T> other);
+
+        /// <summary>
+        /// Gets the values as sorted set
+        /// </summary>
+        /// <returns></returns>
+        SortedMultiSet<T> GetSortedValues();
+
+        /// <summary>
+        /// Returns total number of values in the set
+        /// </summary>
+        long Count { get; }
+    }
+
+    internal class SimpleTopKState<T> : ITopKState<T>
+    {
+        private SortedMultiSet<T> values;
+
+        public SimpleTopKState(Func<SortedDictionary<T, long>> generator)
+        {
+            this.values = new SortedMultiSet<T>(generator);
+        }
+
+        public long Count => this.values.TotalCount;
+
+        public ITopKState<T> Add(T input, long timestamp)
+        {
+            this.values.Add(input);
+            return this;
+        }
+
+        public SortedMultiSet<T> GetSortedValues() => this.values;
+
+        public ITopKState<T> Remove(T input, long timestamp)
+        {
+            this.values.Remove(input);
+            return this;
+        }
+
+        public ITopKState<T> RemoveAll(ITopKState<T> other)
+        {
+            this.values.RemoveAll(other.GetSortedValues());
+            return this;
+        }
+    }
+
+    internal class HoppingTopKState<T> : ITopKState<T>
+    {
+        public long currentTimestamp;
+
+        public CircularBuffer<ValueTuple<long, SortedMultiSet<T>>> previousValues;
+        public SortedMultiSet<T> currentValues;
+
+        public int k;
+
+        public Comparison<T> rankComparer;
+        private Func<SortedDictionary<T, long>> generator;
+        private ItemAndCount<T> minValue; // The minimum threshold value in TopK
+
+        public HoppingTopKState(int k, Comparison<T> rankComparer, int hoppingWindowSize, Func<SortedDictionary<T, long>> generator)
+        {
+            this.k = k;
+            this.rankComparer = rankComparer;
+            this.currentValues = new SortedMultiSet<T>(generator);
+            this.previousValues = new CircularBuffer<ValueTuple<long, SortedMultiSet<T>>>(hoppingWindowSize);
+            this.generator = generator;
+        }
+
+        public ITopKState<T> Add(T input, long timestamp)
+        {
+            // Verify that input is added in non-decreasing order
+            if (timestamp < this.currentTimestamp)
+            {
+                throw new ArgumentException("Invalid timestamp");
+            }
+
+            // First entry in new hop window, just add the value
+            if (timestamp > this.currentTimestamp)
+            {
+                MergeCurrentToPrevious();
+                this.currentTimestamp = timestamp;
+                this.currentValues.Add(input);
+                this.minValue = new ItemAndCount<T>(input, 1);
+                return this;
+            }
+
+            // these are subsequent entries
+            int compare = rankComparer(input, this.minValue.Item);
+
+            if (this.currentValues.TotalCount < this.k) // if we have not reached k yet, add it
+            {
+                if (compare > 0)
+                    this.minValue = new ItemAndCount<T>(input, 1);
+                else if (compare == 0)
+                    this.minValue.Count++;
+
+                this.currentValues.Add(input);
+                return this;
+            }
+            else if (compare > 0) // We have reached k and new input is smaller than minimum
+            {
+                return this;
+            }
+            else if (compare == 0) // We have reached k and new input is equal to the minimum
+            {
+                this.currentValues.Add(input); // add to get ties
+                this.minValue.Count++;
+                return this;
+            }
+            else // The new item is less than minValue, so we need to remove some entries to make place for the new entry
+            {
+                this.currentValues.Add(input);
+                var toRemove = this.currentValues.TotalCount - this.k;
+                if (toRemove >= minValue.Count)
+                {
+                    this.currentValues.RemoveAll(this.minValue.Item);
+                    this.minValue = this.currentValues.GetMinItem();
+                }
+                return this;
+            }
+        }
+
+        public ITopKState<T> Remove(T input, long timestamp)
+        {
+            throw new NotImplementedException("Cannot remove single elements from this state");
+        }
+
+        public ITopKState<T> RemoveAll(ITopKState<T> other)
+        {
+            if (other.Count != 0)
+            {
+                if (other is HoppingTopKState<T> otherTopK)
+                {
+                    if (otherTopK.currentTimestamp > this.currentTimestamp)
+                    {
+                        throw new ArgumentException("Cannot remove entries with current or future timestamp");
+                    }
+                    else if (otherTopK.currentTimestamp == this.currentTimestamp)
+                    {
+                        if (this.currentValues.TotalCount != otherTopK.currentValues.TotalCount)
+                            throw new InvalidOperationException("Invalid removal");
+
+                        this.currentValues.Clear();
+                        this.previousValues.Clear();
+                    }
+                    else
+                    {
+                        while (this.previousValues.Count > 0)
+                        {
+                            var first = this.previousValues.PeekFirst();
+
+                            if (first.Item1 > otherTopK.currentTimestamp)
+                            {
+                                break;
+                            }
+
+                            if (first.Item1 == otherTopK.currentTimestamp &&
+                                first.Item2.TotalCount != otherTopK.currentValues.TotalCount)
+                                throw new InvalidOperationException("Invalid removal");
+
+                            this.previousValues.Dequeue();
+                        }
+                    }
+                }
+                else
+                {
+                    throw new InvalidOperationException("Cannot remove non-HoppingTopKState from HoppingTopKState");
+                }
+            }
+            return this;
+        }
+
+        // This function merges the current values to previous and is expensive
+        // Currently it is only called by ComputeResult
+        public SortedMultiSet<T> GetSortedValues()
+        {
+            var sortedMultiset = new SortedMultiSet<T>(generator);
+
+            foreach (var dictItem in this.previousValues.Iterate())
+            {
+                sortedMultiset.AddAll(dictItem.Item2);
+            }
+            sortedMultiset.AddAll(this.currentValues);
+
+            return sortedMultiset;
+        }
+
+        private void MergeCurrentToPrevious()
+        {
+            if (!this.currentValues.IsEmpty)
+            {
+                var newEntry = ValueTuple.Create(this.currentTimestamp, this.currentValues);
+                this.previousValues.Enqueue(ref newEntry);
+                this.currentValues = new SortedMultiSet<T>(generator);
+            }
+        }
+
+        public long Count => this.currentValues.TotalCount + this.previousValues.Iterate().Sum(e => e.Item2.TotalCount);
+    }
+}
diff --git a/Sources/Core/Microsoft.StreamProcessing/Collections/CircularBuffer.cs b/Sources/Core/Microsoft.StreamProcessing/Collections/CircularBuffer.cs
index 3df34156c..fff6bf7be 100644
--- a/Sources/Core/Microsoft.StreamProcessing/Collections/CircularBuffer.cs
+++ b/Sources/Core/Microsoft.StreamProcessing/Collections/CircularBuffer.cs
@@ -105,6 +105,13 @@ public T Dequeue()
         [EditorBrowsable(EditorBrowsableState.Never)]
         public bool IsEmpty() => this.head == this.tail;
 
+        /// <summary>
+        /// Removes alll elements from the list - do not use directly.
+        /// </summary>
+        [MethodImpl(MethodImplOptions.AggressiveInlining)]
+        [EditorBrowsable(EditorBrowsableState.Never)]
+        public void Clear() => this.head = this.tail = 0;
+
         /// <summary>
         /// Currently for internal use only - do not use directly.
         /// </summary>
@@ -260,6 +267,17 @@ public IEnumerator<T> GetEnumerator()
             }
         }
 
+        /// <summary>
+        /// Currently for internal use only - do not use directly.
+        /// </summary>
+        [EditorBrowsable(EditorBrowsableState.Never)]
+        public void Clear()
+        {
+            this.tail = this.head = this.buffers.First;
+            this.head.Value.Clear();
+            this.Count = 0;
+        }
+
         System.Collections.IEnumerator System.Collections.IEnumerable.GetEnumerator() => GetEnumerator();
     }
 }
diff --git a/Sources/Core/Microsoft.StreamProcessing/Collections/SortedMultiSet.cs b/Sources/Core/Microsoft.StreamProcessing/Collections/SortedMultiSet.cs
index 84ff1aeaf..412baf5a5 100644
--- a/Sources/Core/Microsoft.StreamProcessing/Collections/SortedMultiSet.cs
+++ b/Sources/Core/Microsoft.StreamProcessing/Collections/SortedMultiSet.cs
@@ -182,6 +182,25 @@ public SortedMultiSet<T> Remove(T key)
             return this;
         }
 
+        /// <summary>
+        /// Remove all items matching the key
+        /// </summary>
+        /// <param name="key">The element that should be removed from the current object.</param>
+        /// <returns>A reference to the current Sorted Multiset to allow for functional composition.</returns>
+        [MethodImpl(MethodImplOptions.AggressiveInlining)]
+        [EditorBrowsable(EditorBrowsableState.Never)]
+        public SortedMultiSet<T> RemoveAll(T key)
+        {
+            this.Elements.TryGetValue(key, out long keyCount);
+            if (keyCount == 0)
+                throw new ArgumentException("Attempting to remove element not in set.");
+
+            this.Elements.Remove(key);
+
+            this.count -= keyCount;
+            return this;
+        }
+
         /// <summary>
         /// Remove the given element from the current Sorted Multiset the given number of times.
         /// </summary>
@@ -221,6 +240,21 @@ public SortedMultiSet<T> RemoveAll(SortedMultiSet<T> set)
             return this;
         }
 
+        /// <summary>
+        /// Remove all of the elements in the given set from the current Sorted Multiset, inluding multiplicity.
+        /// </summary>
+        /// <param name="itemAndCount">The Sorted Multiset containing the elements that should be removed from the current object.</param>
+        /// <returns>A reference to the current Sorted Multiset to allow for functional composition.</returns>
+        [EditorBrowsable(EditorBrowsableState.Never)]
+        public SortedMultiSet<T> RemoveAll(IEnumerable<ItemAndCount<T>> itemAndCount)
+        {
+            Contract.Requires(itemAndCount != null);
+            foreach (var keyAndCount in itemAndCount)
+                Remove(keyAndCount.Item, keyAndCount.Count);
+
+            return this;
+        }
+
         /// <summary>
         /// Clear the contents of the Sorted Multiset.
         /// </summary>
@@ -267,5 +301,52 @@ public IEnumerable<T> GetEnumerable()
                     yield return keyAndCount.Key;
             }
         }
+
+        /// <summary>
+        /// Get all elements, including multiplicity, of the Sorted Multiset in sort order.
+        /// </summary>
+        /// <returns>An object that enumerates all of the elements in the Sorted Multiset, including multiplicity.</returns>
+        [EditorBrowsable(EditorBrowsableState.Never)]
+        public IEnumerable<ItemAndCount<T>> GetItemCountEnumerable()
+        {
+            foreach (var keyAndCount in this.Elements)
+            {
+                yield return new ItemAndCount<T>(keyAndCount.Key, keyAndCount.Value);
+            }
+        }
+
+        /// <summary>
+        /// Get the smallest element (and its number of occurances) in the collection
+        /// </summary>
+        /// <returns>The smallest element and its count</returns>
+        [EditorBrowsable(EditorBrowsableState.Never)]
+        public ItemAndCount<T> GetMinItem()
+        {
+            var minItem = this.Elements.LastOrDefault();
+            return new ItemAndCount<T>(minItem.Key, minItem.Value);
+        }
+    }
+
+    /// <summary>
+    /// Represents an Item and its count used in MultiSet
+    /// </summary>
+    /// <typeparam name="T"></typeparam>
+    public struct ItemAndCount<T>
+    {
+        internal ItemAndCount(T item, long count)
+        {
+            Item = item;
+            Count = count;
+        }
+
+        /// <summary>
+        /// Item type used in Multiset
+        /// </summary>
+        public readonly T Item;
+
+        /// <summary>
+        /// Count of items with same value
+        /// </summary>
+        public long Count { get; set; }
     }
 }
diff --git a/Sources/Core/Microsoft.StreamProcessing/Windows/Window.cs b/Sources/Core/Microsoft.StreamProcessing/Windows/Window.cs
index 198521763..7ee4a74ea 100644
--- a/Sources/Core/Microsoft.StreamProcessing/Windows/Window.cs
+++ b/Sources/Core/Microsoft.StreamProcessing/Windows/Window.cs
@@ -162,25 +162,31 @@ public IAggregate<TSource, MinMaxState<TValue>, TValue> Max<TValue>(Expression<F
         /// <summary>
         /// Computes a time-sensitive top-k aggregate using snapshot semantics based on a key selector.
         /// </summary>
-        public IAggregate<TSource, SortedMultiSet<TSource>, List<RankedEvent<TSource>>> TopK<TOrderValue>(Expression<Func<TSource, TOrderValue>> orderer, int k)
+        public IAggregate<TSource, ITopKState<TSource>, List<RankedEvent<TSource>>> TopK<TOrderValue>(Expression<Func<TSource, TOrderValue>> orderer, int k)
         {
             Invariant.IsNotNull(orderer, nameof(orderer));
             Invariant.IsPositive(k, nameof(k));
             var orderComparer = ComparerExpression<TOrderValue>.Default.TransformInput(orderer);
-            var aggregate = new TopKAggregate<TSource>(k, orderComparer, this.Properties.QueryContainer);
+            bool isHoppingWindow = !Properties.IsTumbling && Properties.IsConstantDuration && Properties.ConstantDurationLength.HasValue &&
+                                   Properties.IsConstantHop && Properties.ConstantHopLength.HasValue && (Properties.ConstantDurationLength.Value % Properties.ConstantHopLength.Value) == 0;
+            long hoppingWindowSize = isHoppingWindow ? (Properties.ConstantDurationLength.Value / Properties.ConstantHopLength.Value) : 0;
+            var aggregate = new TopKAggregate<TSource>(k, orderComparer, this.Properties.QueryContainer, hoppingWindowSize);
             return aggregate.SkipNulls().ApplyFilter(this.Filter);
         }
 
         /// <summary>
         /// Computes a time-sensitive top-k aggregate using snapshot semantics based on a key selector with the provided ordering comparer.
         /// </summary>
-        public IAggregate<TSource, SortedMultiSet<TSource>, List<RankedEvent<TSource>>> TopK<TOrderValue>(Expression<Func<TSource, TOrderValue>> orderer, IComparerExpression<TOrderValue> comparer, int k)
+        public IAggregate<TSource, ITopKState<TSource>, List<RankedEvent<TSource>>> TopK<TOrderValue>(Expression<Func<TSource, TOrderValue>> orderer, IComparerExpression<TOrderValue> comparer, int k)
         {
             Invariant.IsNotNull(orderer, nameof(orderer));
             Invariant.IsNotNull(comparer, nameof(comparer));
             Invariant.IsPositive(k, nameof(k));
             var orderComparer = comparer.TransformInput(orderer);
-            var aggregate = new TopKAggregate<TSource>(k, orderComparer, this.Properties.QueryContainer);
+            bool isHoppingWindow = !Properties.IsTumbling && Properties.IsConstantDuration && Properties.ConstantDurationLength.HasValue &&
+                                   Properties.IsConstantHop && Properties.ConstantHopLength.HasValue && (Properties.ConstantHopLength.Value % Properties.ConstantDurationLength.Value) == 0;
+            long hoppingWindowSize = isHoppingWindow ? (Properties.ConstantDurationLength.Value / Properties.ConstantHopLength.Value) : 0;
+            var aggregate = new TopKAggregate<TSource>(k, orderComparer, this.Properties.QueryContainer, hoppingWindowSize);
             return aggregate.SkipNulls().ApplyFilter(this.Filter);
         }
 
diff --git a/Sources/Test/SimpleTesting/Aggregates/HoppingWindowTopKAggregate.cs b/Sources/Test/SimpleTesting/Aggregates/HoppingWindowTopKAggregate.cs
new file mode 100644
index 000000000..2cfa02b91
--- /dev/null
+++ b/Sources/Test/SimpleTesting/Aggregates/HoppingWindowTopKAggregate.cs
@@ -0,0 +1,220 @@
+// *********************************************************************
+// Copyright (c) Microsoft Corporation.  All rights reserved.
+// Licensed under the MIT License
+// *********************************************************************
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using Microsoft.StreamProcessing;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+
+namespace SimpleTesting
+{
+    [TestClass]
+    public class HoppingWindowTopKAggregate : TestWithConfigSettingsAndMemoryLeakDetection
+    {
+        private const long HopSize = 10;
+        private const long WindowSize = 4 * HopSize;
+        private const int K = 3;
+
+        private StreamEvent<int> EndEvent = StreamEvent.CreatePunctuation<int>(StreamEvent.InfinitySyncTime);
+        private Random random = new Random(Seed: (int)DateTime.UtcNow.Ticks);
+
+        [TestMethod, TestCategory("Gated")]
+        public void TestHoppingWindowTopKAggregateSimple()
+        {
+            var input = new[]
+            {
+                StreamEvent.CreateStart(1, 10),
+                StreamEvent.CreateStart(3, 20),
+                StreamEvent.CreateStart(6, 10),
+                StreamEvent.CreateStart(8, 30),
+                StreamEvent.CreateStart(12, 20),
+                StreamEvent.CreateStart(18, 10),
+                EndEvent
+            };
+
+            var output = input.ToStreamable().HoppingWindowLifetime(10, 5).TopK(K);
+            var outputValues = output.Select(re => re.Select(t => t.Payload).ToArray())
+                .ToStreamEventArray().Where(e => e.IsData).ToArray();
+
+            var expected = new[]
+            {
+                StreamEvent.CreateStart(5, new[] { 20, 10 }),
+                StreamEvent.CreateEnd(10, 5, new[] { 20, 10 }),
+                StreamEvent.CreateStart(10, new[] { 30, 20, 10, 10 }),
+                StreamEvent.CreateEnd(15, 10, new[] { 30, 20, 10, 10 }),
+                StreamEvent.CreateStart(15, new[] { 30, 20, 10 }),
+                StreamEvent.CreateEnd(20, 15, new[] { 30, 20, 10 }),
+                StreamEvent.CreateStart(20, new[] { 20, 10 }),
+                StreamEvent.CreateEnd(25, 20, new[] { 20, 10 }),
+                StreamEvent.CreateStart(25, new[] { 10 }),
+                StreamEvent.CreateEnd(30, 25, new[] { 10 }),
+            };
+
+            Assert.AreEqual(expected.Length, outputValues.Length);
+
+            for (int i = 0; i < expected.Length; i++)
+            {
+                Assert.AreEqual(expected[i].StartTime, outputValues[i].StartTime);
+                Assert.AreEqual(expected[i].EndTime, outputValues[i].EndTime);
+                Assert.AreEqual(expected[i].Payload.Length, outputValues[i].Payload.Length);
+                for (int j = 0; j < expected[i].Payload.Length; j++)
+                    Assert.AreEqual(expected[i].Payload[j], outputValues[i].Payload[j]);
+            }
+        }
+
+        [TestMethod, TestCategory("Gated")]
+        public void TestHoppingWindowTopKAggregateRandomDistribution()
+        {
+            // Distribution: [1,2,3,4,5] hops : 10% each, Closely-Spaced: 50%
+            GenerateDataAndTestInput(
+                numValues: 1000,
+                valueGenerator: v => random.Next(100, 110),
+                distanceGenerator: () =>
+                {
+                    var hopType = random.Next(1, 11);
+                    return (hopType < 6) ? (hopType * HopSize) : (hopType - 5);
+                });
+        }
+
+        [TestMethod, TestCategory("Gated")]
+        public void TestHoppingWindowTopKAggregateUnaligned()
+        {
+            GenerateDataAndTestInput(
+                numValues: 1000,
+                valueGenerator: v => random.Next(100, 110),
+                distanceGenerator: () => 10,
+                windowSize: 27);
+
+            GenerateDataAndTestInput(
+                numValues: 1000,
+                valueGenerator: v => random.Next(100, 110),
+                distanceGenerator: () => 10,
+                windowSize: 17);
+        }
+
+        [TestMethod, TestCategory("Gated")]
+        public void TestHoppingWindowTopKAggregateAllIncreasing()
+        {
+            GenerateDataAndTestInput(
+                numValues: 1000,
+                valueGenerator: v => v + random.Next(1, 11),
+                distanceGenerator: () => HopSize);
+        }
+
+        [TestMethod, TestCategory("Gated")]
+        public void TestHoppingWindowTopKAggregateAllDecreasing()
+        {
+            GenerateDataAndTestInput(
+                numValues: 1000,
+                valueGenerator: v => (v == 0) ? 10000 : v - random.Next(1, 11),
+                distanceGenerator: () => HopSize);
+        }
+
+        [TestMethod, TestCategory("Gated")]
+        public void TestHoppingWindowTopKAggregateCyclingValues()
+        {
+            // Test values 1 -> 2 -> 3 -> 1, and run for 1/2, 1, 2, 4, 8 intervals of hops
+            for (long distance = HopSize / 2; distance <= HopSize * 8; distance *= 2)
+            {
+                GenerateDataAndTestInput(
+                    numValues: 10,
+                    valueGenerator: v => (v % 3) + 1,
+                    distanceGenerator: () => distance);
+            }
+        }
+
+        private void GenerateDataAndTestInput(
+            int numValues,
+            Func<int, int> valueGenerator,
+            Func<long> distanceGenerator,
+            long windowSize = WindowSize)
+        {
+            var input = new List<StreamEvent<int>>();
+            long maxStartTime = 0;
+
+            long startTime = 0;
+            int value = 100;
+            for (int i = 0; i < numValues; i++)
+            {
+                startTime += distanceGenerator();
+                value = valueGenerator(value);
+                input.Add(StreamEvent.CreateStart(startTime, value));
+                maxStartTime = Math.Max(maxStartTime, startTime);
+            }
+            input.Add(EndEvent);
+
+            TestHoppingWindowTopKAggregateInternal(input, maxStartTime, windowSize);
+        }
+
+        private void TestHoppingWindowTopKAggregateInternal(IEnumerable<StreamEvent<int>> streamEvents, long maxStartTime, long windowSize)
+        {
+            var output = streamEvents.ToStreamable().HoppingWindowLifetime(windowSize, HopSize).TopK(K);
+
+            var correct = new List<StreamEvent<List<int>>>();
+
+            maxStartTime += windowSize;
+            for (long startTime = 0; startTime < maxStartTime; startTime += HopSize)
+            {
+                var eventsInWindow = streamEvents.Where(x => x.StartTime > (startTime - windowSize) && x.StartTime <= startTime);
+
+                if (!eventsInWindow.Any())
+                    continue;
+
+                var dataInWindow = eventsInWindow.Select(se => se.Payload).ToList();
+                dataInWindow.Sort((x, y) => y.CompareTo(x));
+
+                var last = dataInWindow[Math.Min(K, dataInWindow.Count) - 1];
+                var topKInWindow = dataInWindow.TakeWhile(e => e >= last).ToList();
+
+                correct.Add(StreamEvent.CreateStart(startTime, topKInWindow));
+                correct.Add(StreamEvent.CreateEnd(startTime + HopSize, startTime, topKInWindow));
+            }
+
+            var output2 = output.Select(e => e.Select(re => re.Payload).ToList());
+
+            var expected = NormalizeToInterval(correct).ToList();
+            var actual = NormalizeToInterval(output2.ToStreamEventArray()).ToList();
+
+            Assert.AreEqual(expected.Count, actual.Count);
+
+            for (int i = 0; i < expected.Count; i++)
+            {
+                Assert.AreEqual(expected[i].Kind, actual[i].Kind);
+                Assert.AreEqual(expected[i].StartTime, actual[i].StartTime);
+                Assert.AreEqual(expected[i].EndTime, actual[i].EndTime);
+                CollectionAssert.AreEquivalent(expected[i].Payload, actual[i].Payload);
+            }
+        }
+
+        private IEnumerable<StreamEvent<List<int>>> NormalizeToInterval(IEnumerable<StreamEvent<List<int>>> streamEvents)
+        {
+            var result = new List<StreamEvent<List<int>>>();
+
+            var endEvents = streamEvents.Where(se => se.Kind == StreamEventKind.End || se.Kind == StreamEventKind.Interval);
+
+            if (!endEvents.Any())
+                return result;
+
+            var firstEvent = endEvents.First();
+            StreamEvent<List<int>> curInterval = StreamEvent.CreateInterval(firstEvent.StartTime, firstEvent.EndTime, firstEvent.Payload);
+
+            foreach (var streamEvent in endEvents.Skip(1))
+            {
+                if ((streamEvent.StartTime == curInterval.EndTime || streamEvent.Kind == StreamEventKind.Interval) && // Merge into current interval if payload is same
+                    Enumerable.SequenceEqual(streamEvent.Payload, curInterval.Payload))
+                {
+                    curInterval.OtherTime = streamEvent.EndTime;
+                }
+                else
+                {
+                    result.Add(curInterval);
+                    curInterval = StreamEvent.CreateInterval(streamEvent.StartTime, streamEvent.EndTime, streamEvent.Payload); ;
+                }
+            }
+            result.Add(curInterval);
+            return result;
+        }
+    }
+}
diff --git a/Sources/Test/SimpleTesting/Program.cs b/Sources/Test/SimpleTesting/Program.cs
index e1b97e1e9..80f4397f2 100644
--- a/Sources/Test/SimpleTesting/Program.cs
+++ b/Sources/Test/SimpleTesting/Program.cs
@@ -49,7 +49,7 @@ public static IStreamable<Empty, TPayload> ToCleanStreamable<TPayload>(this Stre
             return input.OrderBy(v => v.SyncTime).ToArray().ToStreamable();
         }
 
-        public static IStreamable<Empty, TPayload> ToStreamable<TPayload>(this StreamEvent<TPayload>[] input)
+        public static IStreamable<Empty, TPayload> ToStreamable<TPayload>(this IEnumerable<StreamEvent<TPayload>> input)
         {
             Invariant.IsNotNull(input, "input");
 
diff --git a/Sources/Test/SimpleTesting/SimpleTesting.csproj b/Sources/Test/SimpleTesting/SimpleTesting.csproj
index aedf068f0..93c7c8e97 100644
--- a/Sources/Test/SimpleTesting/SimpleTesting.csproj
+++ b/Sources/Test/SimpleTesting/SimpleTesting.csproj
@@ -63,6 +63,7 @@
     <Compile Include="Aggregates\CompoundAggregate.cs" />
     <Compile Include="Aggregates\CountAggregate.cs" />
     <Compile Include="Aggregates\GroupAggregate.cs" />
+    <Compile Include="Aggregates\HoppingWindowTopKAggregate.cs" />
     <Compile Include="Aggregates\MaxAggregate.cs" />
     <Compile Include="Aggregates\MinAggregate.cs" />
     <Compile Include="Aggregates\PercentileAggregates.cs" />