Github user tillrohrmann commented on a diff in the pull request: https://github.com/apache/flink/pull/6062#discussion_r191447784 --- Diff: flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/InternalTimerHeapTest.java --- @@ -0,0 +1,470 @@ +/* + * 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.flink.streaming.api.operators; + +import org.apache.flink.runtime.state.KeyGroupRange; +import org.apache.flink.runtime.state.VoidNamespace; +import org.apache.flink.util.Preconditions; + +import org.junit.Assert; +import org.junit.Test; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashSet; +import java.util.Iterator; +import java.util.List; +import java.util.NoSuchElementException; +import java.util.Set; +import java.util.concurrent.ThreadLocalRandom; +import java.util.function.BiFunction; +import java.util.function.Function; + +/** + * Tests for {@link InternalTimerHeap}. + */ +public class InternalTimerHeapTest { + + private static final KeyGroupRange KEY_GROUP_RANGE = new KeyGroupRange(0, 1); + + private static void insertRandomTimers( + InternalTimerHeap<Integer, VoidNamespace> timerPriorityQueue, + int count) { + insertRandomTimers(timerPriorityQueue, null, count); + } + + private static void insertRandomTimers( + InternalTimerHeap<Integer, VoidNamespace> timerPriorityQueue, + Set<TimerHeapInternalTimer<Integer, VoidNamespace>> checkSet, + int count) { + + ThreadLocalRandom localRandom = ThreadLocalRandom.current(); + + for (int i = 0; i < count; ++i) { + TimerHeapInternalTimer<Integer, VoidNamespace> timer = + new TimerHeapInternalTimer<>(localRandom.nextLong(), i, VoidNamespace.INSTANCE); + if (checkSet != null) { + Preconditions.checkState(checkSet.add(timer)); + } + Assert.assertTrue(timerPriorityQueue.add(timer)); + } + } + + private static InternalTimerHeap<Integer, VoidNamespace> newPriorityQueue(int initialCapacity) { + return new InternalTimerHeap<>( + initialCapacity, + KEY_GROUP_RANGE, + KEY_GROUP_RANGE.getNumberOfKeyGroups()); + } + + @Test + public void testCombined() { + final int initialCapacity = 4; + final int testSize = 1000; + InternalTimerHeap<Integer, VoidNamespace> timerPriorityQueue = newPriorityQueue(initialCapacity); + HashSet<TimerHeapInternalTimer<Integer, VoidNamespace>> checkSet = new HashSet<>(testSize); + + insertRandomTimers(timerPriorityQueue, checkSet, testSize); + + long lastTimestamp = Long.MIN_VALUE; + int lastSize = timerPriorityQueue.size(); + Assert.assertEquals(testSize, lastSize); + TimerHeapInternalTimer<Integer, VoidNamespace> timer; + while ((timer = timerPriorityQueue.peek()) != null) { + Assert.assertFalse(timerPriorityQueue.isEmpty()); + Assert.assertEquals(lastSize, timerPriorityQueue.size()); + Assert.assertEquals(timer, timerPriorityQueue.poll()); + Assert.assertTrue(checkSet.remove(timer)); + Assert.assertTrue(timer.getTimestamp() >= lastTimestamp); + lastTimestamp = timer.getTimestamp(); + --lastSize; + } + + Assert.assertTrue(timerPriorityQueue.isEmpty()); + Assert.assertEquals(0, timerPriorityQueue.size()); + Assert.assertEquals(0, checkSet.size()); + } + + @Test + public void testAdd() { + testAddOfferCommon(InternalTimerHeap<Integer, VoidNamespace>::add); + } + + @Test + public void testOffer() { + testAddOfferCommon(InternalTimerHeap<Integer, VoidNamespace>::offer); --- End diff -- What's the difference between `add` and `offer`? Aren't these effectively the same methods?
---