You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2015/09/22 13:25:49 UTC
[1/3] flink git commit: [FLINK-2683] [FLINK-2682] [runtime] Add
dedicated operator for aligned processing time windows.
Repository: flink
Updated Branches:
refs/heads/master 64e1dc6fe -> b8f58fab5
http://git-wip-us.apache.org/repos/asf/flink/blob/b8f58fab/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/runtime/operators/windows/AggregatingAlignedProcessingTimeWindowOperatorTest.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/runtime/operators/windows/AggregatingAlignedProcessingTimeWindowOperatorTest.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/runtime/operators/windows/AggregatingAlignedProcessingTimeWindowOperatorTest.java
new file mode 100644
index 0000000..464df32
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/runtime/operators/windows/AggregatingAlignedProcessingTimeWindowOperatorTest.java
@@ -0,0 +1,528 @@
+/*
+ * 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.runtime.operators.windows;
+
+import org.apache.flink.api.common.functions.ReduceFunction;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.runtime.operators.TriggerTimer;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.StreamingRuntimeContext;
+
+import org.junit.After;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+@SuppressWarnings("serial")
+public class AggregatingAlignedProcessingTimeWindowOperatorTest {
+
+ @SuppressWarnings("unchecked")
+ private final ReduceFunction<String> mockFunction = mock(ReduceFunction.class);
+
+ @SuppressWarnings("unchecked")
+ private final KeySelector<String, String> mockKeySelector = mock(KeySelector.class);
+
+ private final KeySelector<Integer, Integer> identitySelector = new KeySelector<Integer, Integer>() {
+ @Override
+ public Integer getKey(Integer value) {
+ return value;
+ }
+ };
+
+ private final ReduceFunction<Integer> sumFunction = new ReduceFunction<Integer>() {
+ @Override
+ public Integer reduce(Integer value1, Integer value2) {
+ return value1 + value2;
+ }
+ };
+
+ // ------------------------------------------------------------------------
+
+ @After
+ public void checkNoTriggerThreadsRunning() {
+ // make sure that all the threads we trigger are shut down
+ long deadline = System.currentTimeMillis() + 5000;
+ while (TriggerTimer.TRIGGER_THREADS_GROUP.activeCount() > 0 && System.currentTimeMillis() < deadline) {
+ try {
+ Thread.sleep(10);
+ }
+ catch (InterruptedException ignored) {}
+ }
+
+ assertTrue("Not all trigger threads where properly shut down",
+ TriggerTimer.TRIGGER_THREADS_GROUP.activeCount() == 0);
+ }
+
+ // ------------------------------------------------------------------------
+
+ @Test
+ public void testInvalidParameters() {
+ try {
+ assertInvalidParameter(-1L, -1L);
+ assertInvalidParameter(10000L, -1L);
+ assertInvalidParameter(-1L, 1000L);
+ assertInvalidParameter(1000L, 2000L);
+
+ // actual internal slide is too low here:
+ assertInvalidParameter(1000L, 999L);
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testWindowSizeAndSlide() {
+ try {
+ AbstractAlignedProcessingTimeWindowOperator<String, String, String> op;
+
+ op = new AggregatingProcessingTimeWindowOperator<>(mockFunction, mockKeySelector, 5000, 1000);
+ assertEquals(5000, op.getWindowSize());
+ assertEquals(1000, op.getWindowSlide());
+ assertEquals(1000, op.getPaneSize());
+ assertEquals(5, op.getNumPanesPerWindow());
+
+ op = new AggregatingProcessingTimeWindowOperator<>(mockFunction, mockKeySelector, 1000, 1000);
+ assertEquals(1000, op.getWindowSize());
+ assertEquals(1000, op.getWindowSlide());
+ assertEquals(1000, op.getPaneSize());
+ assertEquals(1, op.getNumPanesPerWindow());
+
+ op = new AggregatingProcessingTimeWindowOperator<>(mockFunction, mockKeySelector, 1500, 1000);
+ assertEquals(1500, op.getWindowSize());
+ assertEquals(1000, op.getWindowSlide());
+ assertEquals(500, op.getPaneSize());
+ assertEquals(3, op.getNumPanesPerWindow());
+
+ op = new AggregatingProcessingTimeWindowOperator<>(mockFunction, mockKeySelector, 1200, 1100);
+ assertEquals(1200, op.getWindowSize());
+ assertEquals(1100, op.getWindowSlide());
+ assertEquals(100, op.getPaneSize());
+ assertEquals(12, op.getNumPanesPerWindow());
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testWindowTriggerTimeAlignment() {
+ try {
+ @SuppressWarnings("unchecked")
+ final Output<StreamRecord<String>> mockOut = mock(Output.class);
+
+ final StreamingRuntimeContext mockContext = mock(StreamingRuntimeContext.class);
+ when(mockContext.getTaskName()).thenReturn("Test task name");
+
+ AbstractAlignedProcessingTimeWindowOperator<String, String, String> op;
+
+ op = new AggregatingProcessingTimeWindowOperator<>(mockFunction, mockKeySelector, 5000, 1000);
+ op.setup(mockOut, mockContext);
+ op.open(new Configuration());
+ assertTrue(op.getNextSlideTime() % 1000 == 0);
+ assertTrue(op.getNextEvaluationTime() % 1000 == 0);
+ op.dispose();
+
+ op = new AggregatingProcessingTimeWindowOperator<>(mockFunction, mockKeySelector, 1000, 1000);
+ op.setup(mockOut, mockContext);
+ op.open(new Configuration());
+ assertTrue(op.getNextSlideTime() % 1000 == 0);
+ assertTrue(op.getNextEvaluationTime() % 1000 == 0);
+ op.dispose();
+
+ op = new AggregatingProcessingTimeWindowOperator<>(mockFunction, mockKeySelector, 1500, 1000);
+ op.setup(mockOut, mockContext);
+ op.open(new Configuration());
+ assertTrue(op.getNextSlideTime() % 500 == 0);
+ assertTrue(op.getNextEvaluationTime() % 1000 == 0);
+ op.dispose();
+
+ op = new AggregatingProcessingTimeWindowOperator<>(mockFunction, mockKeySelector, 1200, 1100);
+ op.setup(mockOut, mockContext);
+ op.open(new Configuration());
+ assertTrue(op.getNextSlideTime() % 100 == 0);
+ assertTrue(op.getNextEvaluationTime() % 1100 == 0);
+ op.dispose();
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testTumblingWindowUniqueElements() {
+ try {
+ final int windowSize = 50;
+ final CollectingOutput<Integer> out = new CollectingOutput<>(windowSize);
+
+ final StreamingRuntimeContext mockContext = mock(StreamingRuntimeContext.class);
+ when(mockContext.getTaskName()).thenReturn("Test task name");
+
+ AggregatingProcessingTimeWindowOperator<Integer, Integer> op =
+ new AggregatingProcessingTimeWindowOperator<>(
+ sumFunction, identitySelector, windowSize, windowSize);
+
+ op.setup(out, mockContext);
+ op.open(new Configuration());
+
+ final int numElements = 1000;
+
+ for (int i = 0; i < numElements; i++) {
+ op.processElement(new StreamRecord<Integer>(i));
+ Thread.sleep(1);
+ }
+
+ op.close();
+ op.dispose();
+
+ // get and verify the result
+ List<Integer> result = out.getElements();
+ assertEquals(numElements, result.size());
+
+ Collections.sort(result);
+ for (int i = 0; i < numElements; i++) {
+ assertEquals(i, result.get(i).intValue());
+ }
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testTumblingWindowDuplicateElements() {
+ try {
+ final int windowSize = 50;
+ final CollectingOutput<Integer> out = new CollectingOutput<>(windowSize);
+
+ final StreamingRuntimeContext mockContext = mock(StreamingRuntimeContext.class);
+ when(mockContext.getTaskName()).thenReturn("Test task name");
+
+ AggregatingProcessingTimeWindowOperator<Integer, Integer> op =
+ new AggregatingProcessingTimeWindowOperator<>(
+ sumFunction, identitySelector, windowSize, windowSize);
+
+ op.setup(out, mockContext);
+ op.open(new Configuration());
+
+ final int numWindows = 10;
+
+ long previousNextTime = 0;
+ int window = 1;
+
+ while (window <= numWindows) {
+ long nextTime = op.getNextEvaluationTime();
+ int val = ((int) nextTime) ^ ((int) (nextTime >>> 32));
+
+ op.processElement(new StreamRecord<Integer>(val));
+
+ if (nextTime != previousNextTime) {
+ window++;
+ previousNextTime = nextTime;
+ }
+
+ Thread.sleep(1);
+ }
+
+ op.close();
+ op.dispose();
+
+ List<Integer> result = out.getElements();
+
+ // we have ideally one element per window. we may have more, when we emitted a value into the
+ // successive window (corner case), so we can have twice the number of elements, in the worst case.
+ assertTrue(result.size() >= numWindows && result.size() <= 2 * numWindows);
+
+ // deduplicate for more accurate checks
+ HashSet<Integer> set = new HashSet<>(result);
+ assertTrue(set.size() == 10 || set.size() == 11);
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testSlidingWindow() {
+ try {
+ final CollectingOutput<Integer> out = new CollectingOutput<>(50);
+
+ final StreamingRuntimeContext mockContext = mock(StreamingRuntimeContext.class);
+ when(mockContext.getTaskName()).thenReturn("Test task name");
+
+ // tumbling window that triggers every 20 milliseconds
+ AggregatingProcessingTimeWindowOperator<Integer, Integer> op =
+ new AggregatingProcessingTimeWindowOperator<>(sumFunction, identitySelector, 150, 50);
+
+ op.setup(out, mockContext);
+ op.open(new Configuration());
+
+ final int numElements = 1000;
+
+ for (int i = 0; i < numElements; i++) {
+ op.processElement(new StreamRecord<Integer>(i));
+ Thread.sleep(1);
+ }
+
+ op.close();
+ op.dispose();
+
+ // get and verify the result
+ List<Integer> result = out.getElements();
+
+ // every element can occur between one and three times
+ if (result.size() < numElements || result.size() > 3 * numElements) {
+ System.out.println(result);
+ fail("Wrong number of results: " + result.size());
+ }
+
+ Collections.sort(result);
+ int lastNum = -1;
+ int lastCount = -1;
+
+ for (int num : result) {
+ if (num == lastNum) {
+ lastCount++;
+ assertTrue(lastCount <= 3);
+ }
+ else {
+ lastNum = num;
+ lastCount = 1;
+ }
+ }
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testSlidingWindowSingleElements() {
+ try {
+ final CollectingOutput<Integer> out = new CollectingOutput<>(50);
+
+ final StreamingRuntimeContext mockContext = mock(StreamingRuntimeContext.class);
+ when(mockContext.getTaskName()).thenReturn("Test task name");
+
+ // tumbling window that triggers every 20 milliseconds
+ AggregatingProcessingTimeWindowOperator<Integer, Integer> op =
+ new AggregatingProcessingTimeWindowOperator<>(sumFunction, identitySelector, 150, 50);
+
+ op.setup(out, mockContext);
+ op.open(new Configuration());
+
+ op.processElement(new StreamRecord<Integer>(1));
+ op.processElement(new StreamRecord<Integer>(2));
+
+ // each element should end up in the output three times
+ // wait until the elements have arrived 6 times in the output
+ out.waitForNElements(6, 120000);
+
+ List<Integer> result = out.getElements();
+ assertEquals(6, result.size());
+
+ Collections.sort(result);
+ assertEquals(Arrays.asList(1, 1, 1, 2, 2, 2), result);
+
+ op.close();
+ op.dispose();
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testEmitTrailingDataOnClose() {
+ try {
+ final CollectingOutput<Integer> out = new CollectingOutput<>();
+
+ final StreamingRuntimeContext mockContext = mock(StreamingRuntimeContext.class);
+ when(mockContext.getTaskName()).thenReturn("Test task name");
+
+ // the operator has a window time that is so long that it will not fire in this test
+ final long oneYear = 365L * 24 * 60 * 60 * 1000;
+ AggregatingProcessingTimeWindowOperator<Integer, Integer> op =
+ new AggregatingProcessingTimeWindowOperator<>(sumFunction, identitySelector, oneYear, oneYear);
+
+ op.setup(out, mockContext);
+ op.open(new Configuration());
+
+ List<Integer> data = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
+ for (Integer i : data) {
+ op.processElement(new StreamRecord<Integer>(i));
+ }
+
+ op.close();
+ op.dispose();
+
+ // get and verify the result
+ List<Integer> result = out.getElements();
+ Collections.sort(result);
+ assertEquals(data, result);
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testPropagateExceptionsFromTrigger() {
+ try {
+ final CollectingOutput<Integer> out = new CollectingOutput<>();
+
+ final StreamingRuntimeContext mockContext = mock(StreamingRuntimeContext.class);
+ when(mockContext.getTaskName()).thenReturn("Test task name");
+
+ ReduceFunction<Integer> failingFunction = new FailingFunction(100);
+
+ AggregatingProcessingTimeWindowOperator<Integer, Integer> op =
+ new AggregatingProcessingTimeWindowOperator<>(failingFunction, identitySelector, 200, 50);
+
+ op.setup(out, mockContext);
+ op.open(new Configuration());
+
+ try {
+ long nextWindowTime = op.getNextEvaluationTime();
+ int val = 0;
+ for (int num = 0; num < Integer.MAX_VALUE; num++) {
+ op.processElement(new StreamRecord<Integer>(val++));
+ Thread.sleep(1);
+
+ // when the window has advanced, reset the value, to generate the same values
+ // in the next pane again. This causes the aggregation on trigger to reduce values
+ if (op.getNextEvaluationTime() != nextWindowTime) {
+ nextWindowTime = op.getNextEvaluationTime();
+ val = 0;
+ }
+ }
+ fail("This should really have failed with an exception quite a while ago...");
+ }
+ catch (Exception e) {
+ assertNotNull(e.getCause());
+ assertTrue(e.getCause().getMessage().contains("Artificial Test Exception"));
+ }
+
+ op.dispose();
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testPropagateExceptionsFromProcessElement() {
+ try {
+ final CollectingOutput<Integer> out = new CollectingOutput<>();
+
+ final StreamingRuntimeContext mockContext = mock(StreamingRuntimeContext.class);
+ when(mockContext.getTaskName()).thenReturn("Test task name");
+
+ ReduceFunction<Integer> failingFunction = new FailingFunction(100);
+
+ // the operator has a window time that is so long that it will not fire in this test
+ final long hundredYears = 100L * 365 * 24 * 60 * 60 * 1000;
+ AggregatingProcessingTimeWindowOperator<Integer, Integer> op =
+ new AggregatingProcessingTimeWindowOperator<>(
+ failingFunction, identitySelector, hundredYears, hundredYears);
+
+ op.setup(out, mockContext);
+ op.open(new Configuration());
+
+ for (int i = 0; i < 100; i++) {
+ op.processElement(new StreamRecord<Integer>(1));
+ }
+
+ try {
+ op.processElement(new StreamRecord<Integer>(1));
+ fail("This fail with an exception");
+ }
+ catch (Exception e) {
+ assertTrue(e.getMessage().contains("Artificial Test Exception"));
+ }
+
+ op.dispose();
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ // ------------------------------------------------------------------------
+
+ private void assertInvalidParameter(long windowSize, long windowSlide) {
+ try {
+ new AggregatingProcessingTimeWindowOperator<String, String>(
+ mockFunction, mockKeySelector, windowSize, windowSlide);
+ fail("This should fail with an IllegalArgumentException");
+ }
+ catch (IllegalArgumentException e) {
+ // expected
+ }
+ catch (Exception e) {
+ fail("Wrong exception. Expected IllegalArgumentException but found " + e.getClass().getSimpleName());
+ }
+ }
+
+ // ------------------------------------------------------------------------
+
+ private static class FailingFunction implements ReduceFunction<Integer> {
+
+ private final int failAfterElements;
+
+ private int numElements;
+
+ FailingFunction(int failAfterElements) {
+ this.failAfterElements = failAfterElements;
+ }
+
+ @Override
+ public Integer reduce(Integer value1, Integer value2) throws Exception {
+ numElements++;
+
+ if (numElements >= failAfterElements) {
+ throw new Exception("Artificial Test Exception");
+ }
+
+ return value1 + value2;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/b8f58fab/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/runtime/operators/windows/CollectingOutput.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/runtime/operators/windows/CollectingOutput.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/runtime/operators/windows/CollectingOutput.java
new file mode 100644
index 0000000..9f6858d
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/runtime/operators/windows/CollectingOutput.java
@@ -0,0 +1,80 @@
+/*
+ * 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.runtime.operators.windows;
+
+import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class CollectingOutput<T> implements Output<StreamRecord<T>> {
+
+ private final List<T> elements = new ArrayList<>();
+
+ private final int timeStampModulus;
+
+
+ public CollectingOutput() {
+ this.timeStampModulus = 0;
+ }
+
+ public CollectingOutput(int timeStampModulus) {
+ this.timeStampModulus = timeStampModulus;
+ }
+
+ // ------------------------------------------------------------------------
+
+ public List<T> getElements() {
+ return elements;
+ }
+
+ public void waitForNElements(int n, long timeout) throws InterruptedException {
+ long deadline = System.currentTimeMillis() + timeout;
+ synchronized (elements) {
+ long now;
+ while (elements.size() < n && (now = System.currentTimeMillis()) < deadline) {
+ elements.wait(deadline - now);
+ }
+ }
+ }
+
+ // ------------------------------------------------------------------------
+
+ @Override
+ public void emitWatermark(Watermark mark) {
+ throw new UnsupportedOperationException("the output should not emit watermarks");
+ }
+
+ @Override
+ public void collect(StreamRecord<T> record) {
+ elements.add(record.getValue());
+
+ if (timeStampModulus != 0 && record.getTimestamp() % timeStampModulus != 0) {
+ throw new IllegalArgumentException("Invalid timestamp");
+ }
+ synchronized (elements) {
+ elements.notifyAll();
+ }
+ }
+
+ @Override
+ public void close() {}
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/b8f58fab/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/runtime/operators/windows/KeyMapPutIfAbsentTest.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/runtime/operators/windows/KeyMapPutIfAbsentTest.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/runtime/operators/windows/KeyMapPutIfAbsentTest.java
new file mode 100644
index 0000000..2a9e203
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/runtime/operators/windows/KeyMapPutIfAbsentTest.java
@@ -0,0 +1,121 @@
+/*
+ * 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.runtime.operators.windows;
+
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+public class KeyMapPutIfAbsentTest {
+
+ @Test
+ public void testPutIfAbsentUniqueKeysAndGrowth() {
+ try {
+ KeyMap<Integer, Integer> map = new KeyMap<>();
+ IntegerFactory factory = new IntegerFactory();
+
+ final int numElements = 1000000;
+
+ for (int i = 0; i < numElements; i++) {
+ factory.set(2 * i + 1);
+ map.putIfAbsent(i, factory);
+
+ assertEquals(i+1, map.size());
+ assertTrue(map.getCurrentTableCapacity() > map.size());
+ assertTrue(map.getCurrentTableCapacity() > map.getRehashThreshold());
+ assertTrue(map.size() <= map.getRehashThreshold());
+ }
+
+ assertEquals(numElements, map.size());
+ assertEquals(numElements, map.traverseAndCountElements());
+ assertEquals(1 << 21, map.getCurrentTableCapacity());
+
+ for (int i = 0; i < numElements; i++) {
+ assertEquals(2 * i + 1, map.get(i).intValue());
+ }
+
+ for (int i = numElements - 1; i >= 0; i--) {
+ assertEquals(2 * i + 1, map.get(i).intValue());
+ }
+
+ assertEquals(numElements, map.size());
+ assertEquals(numElements, map.traverseAndCountElements());
+ assertEquals(1 << 21, map.getCurrentTableCapacity());
+ assertTrue(map.getLongestChainLength() <= 7);
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testPutIfAbsentDuplicateKeysAndGrowth() {
+ try {
+ KeyMap<Integer, Integer> map = new KeyMap<>();
+ IntegerFactory factory = new IntegerFactory();
+
+ final int numElements = 1000000;
+
+ for (int i = 0; i < numElements; i++) {
+ int val = 2 * i + 1;
+ factory.set(val);
+ Integer put = map.putIfAbsent(i, factory);
+ assertEquals(val, put.intValue());
+ }
+
+ for (int i = 0; i < numElements; i += 3) {
+ factory.set(2 * i);
+ Integer put = map.putIfAbsent(i, factory);
+ assertEquals(2 * i + 1, put.intValue());
+ }
+
+ for (int i = 0; i < numElements; i++) {
+ assertEquals(2 * i + 1, map.get(i).intValue());
+ }
+
+ assertEquals(numElements, map.size());
+ assertEquals(numElements, map.traverseAndCountElements());
+ assertEquals(1 << 21, map.getCurrentTableCapacity());
+ assertTrue(map.getLongestChainLength() <= 7);
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ // ------------------------------------------------------------------------
+
+ private static class IntegerFactory implements KeyMap.LazyFactory<Integer> {
+
+ private Integer toCreate;
+
+ public void set(Integer toCreate) {
+ this.toCreate = toCreate;
+ }
+
+ @Override
+ public Integer create() {
+ return toCreate;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/b8f58fab/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/runtime/operators/windows/KeyMapPutTest.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/runtime/operators/windows/KeyMapPutTest.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/runtime/operators/windows/KeyMapPutTest.java
new file mode 100644
index 0000000..7335976
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/runtime/operators/windows/KeyMapPutTest.java
@@ -0,0 +1,136 @@
+/*
+ * 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.runtime.operators.windows;
+
+import org.junit.Test;
+
+import java.util.BitSet;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+public class KeyMapPutTest {
+
+ @Test
+ public void testPutUniqueKeysAndGrowth() {
+ try {
+ KeyMap<Integer, Integer> map = new KeyMap<>();
+
+ final int numElements = 1000000;
+
+ for (int i = 0; i < numElements; i++) {
+ map.put(i, 2 * i + 1);
+
+ assertEquals(i+1, map.size());
+ assertTrue(map.getCurrentTableCapacity() > map.size());
+ assertTrue(map.getCurrentTableCapacity() > map.getRehashThreshold());
+ assertTrue(map.size() <= map.getRehashThreshold());
+ }
+
+ assertEquals(numElements, map.size());
+ assertEquals(numElements, map.traverseAndCountElements());
+ assertEquals(1 << 21, map.getCurrentTableCapacity());
+
+ for (int i = 0; i < numElements; i++) {
+ assertEquals(2 * i + 1, map.get(i).intValue());
+ }
+
+ for (int i = numElements - 1; i >= 0; i--) {
+ assertEquals(2 * i + 1, map.get(i).intValue());
+ }
+
+ BitSet bitset = new BitSet();
+ int numContained = 0;
+ for (KeyMap.Entry<Integer, Integer> entry : map) {
+ numContained++;
+
+ assertEquals(entry.getKey() * 2 + 1, entry.getValue().intValue());
+ assertFalse(bitset.get(entry.getKey()));
+ bitset.set(entry.getKey());
+ }
+
+ assertEquals(numElements, numContained);
+ assertEquals(numElements, bitset.cardinality());
+
+
+ assertEquals(numElements, map.size());
+ assertEquals(numElements, map.traverseAndCountElements());
+ assertEquals(1 << 21, map.getCurrentTableCapacity());
+ assertTrue(map.getLongestChainLength() <= 7);
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testPutDuplicateKeysAndGrowth() {
+ try {
+ final KeyMap<Integer, Integer> map = new KeyMap<>();
+ final int numElements = 1000000;
+
+ for (int i = 0; i < numElements; i++) {
+ Integer put = map.put(i, 2*i+1);
+ assertNull(put);
+ }
+
+ for (int i = 0; i < numElements; i += 3) {
+ Integer put = map.put(i, 2*i);
+ assertNotNull(put);
+ assertEquals(2*i+1, put.intValue());
+ }
+
+ for (int i = 0; i < numElements; i++) {
+ int expected = (i % 3 == 0) ? (2*i) : (2*i+1);
+ assertEquals(expected, map.get(i).intValue());
+ }
+
+ assertEquals(numElements, map.size());
+ assertEquals(numElements, map.traverseAndCountElements());
+ assertEquals(1 << 21, map.getCurrentTableCapacity());
+ assertTrue(map.getLongestChainLength() <= 7);
+
+
+ BitSet bitset = new BitSet();
+ int numContained = 0;
+ for (KeyMap.Entry<Integer, Integer> entry : map) {
+ numContained++;
+
+ int key = entry.getKey();
+ int expected = key % 3 == 0 ? (2*key) : (2*key+1);
+
+ assertEquals(expected, entry.getValue().intValue());
+ assertFalse(bitset.get(key));
+ bitset.set(key);
+ }
+
+ assertEquals(numElements, numContained);
+ assertEquals(numElements, bitset.cardinality());
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/b8f58fab/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/runtime/operators/windows/KeyMapTest.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/runtime/operators/windows/KeyMapTest.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/runtime/operators/windows/KeyMapTest.java
new file mode 100644
index 0000000..be71af2
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/runtime/operators/windows/KeyMapTest.java
@@ -0,0 +1,344 @@
+/*
+ * 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.runtime.operators.windows;
+
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Random;
+
+import static org.junit.Assert.*;
+
+public class KeyMapTest {
+
+ @Test
+ public void testInitialSizeComputation() {
+ try {
+ KeyMap<String, String> map;
+
+ map = new KeyMap<>();
+ assertEquals(64, map.getCurrentTableCapacity());
+ assertEquals(6, map.getLog2TableCapacity());
+ assertEquals(24, map.getShift());
+ assertEquals(48, map.getRehashThreshold());
+
+ map = new KeyMap<>(0);
+ assertEquals(64, map.getCurrentTableCapacity());
+ assertEquals(6, map.getLog2TableCapacity());
+ assertEquals(24, map.getShift());
+ assertEquals(48, map.getRehashThreshold());
+
+ map = new KeyMap<>(1);
+ assertEquals(64, map.getCurrentTableCapacity());
+ assertEquals(6, map.getLog2TableCapacity());
+ assertEquals(24, map.getShift());
+ assertEquals(48, map.getRehashThreshold());
+
+ map = new KeyMap<>(9);
+ assertEquals(64, map.getCurrentTableCapacity());
+ assertEquals(6, map.getLog2TableCapacity());
+ assertEquals(24, map.getShift());
+ assertEquals(48, map.getRehashThreshold());
+
+ map = new KeyMap<>(63);
+ assertEquals(64, map.getCurrentTableCapacity());
+ assertEquals(6, map.getLog2TableCapacity());
+ assertEquals(24, map.getShift());
+ assertEquals(48, map.getRehashThreshold());
+
+ map = new KeyMap<>(64);
+ assertEquals(128, map.getCurrentTableCapacity());
+ assertEquals(7, map.getLog2TableCapacity());
+ assertEquals(23, map.getShift());
+ assertEquals(96, map.getRehashThreshold());
+
+ map = new KeyMap<>(500);
+ assertEquals(512, map.getCurrentTableCapacity());
+ assertEquals(9, map.getLog2TableCapacity());
+ assertEquals(21, map.getShift());
+ assertEquals(384, map.getRehashThreshold());
+
+ map = new KeyMap<>(127);
+ assertEquals(128, map.getCurrentTableCapacity());
+ assertEquals(7, map.getLog2TableCapacity());
+ assertEquals(23, map.getShift());
+ assertEquals(96, map.getRehashThreshold());
+
+ // no negative number of elements
+ try {
+ new KeyMap<>(-1);
+ fail("should fail with an exception");
+ }
+ catch (IllegalArgumentException e) {
+ // expected
+ }
+
+ // check integer overflow
+ try {
+ map = new KeyMap<>(0x65715522);
+
+ final int maxCap = Integer.highestOneBit(Integer.MAX_VALUE);
+ assertEquals(Integer.highestOneBit(Integer.MAX_VALUE), map.getCurrentTableCapacity());
+ assertEquals(30, map.getLog2TableCapacity());
+ assertEquals(0, map.getShift());
+ assertEquals(maxCap / 4 * 3, map.getRehashThreshold());
+ }
+ catch (OutOfMemoryError e) {
+ // this may indeed happen in small test setups. we tolerate this in this test
+ }
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testPutAndGetRandom() {
+ try {
+ final KeyMap<Integer, Integer> map = new KeyMap<>();
+ final Random rnd = new Random();
+
+ final long seed = rnd.nextLong();
+ final int numElements = 10000;
+
+ final HashMap<Integer, Integer> groundTruth = new HashMap<>();
+
+ rnd.setSeed(seed);
+ for (int i = 0; i < numElements; i++) {
+ Integer key = rnd.nextInt();
+ Integer value = rnd.nextInt();
+
+ if (rnd.nextBoolean()) {
+ groundTruth.put(key, value);
+ map.put(key, value);
+ }
+ }
+
+ rnd.setSeed(seed);
+ for (int i = 0; i < numElements; i++) {
+ Integer key = rnd.nextInt();
+
+ // skip these, evaluating it is tricky due to duplicates
+ rnd.nextInt();
+ rnd.nextBoolean();
+
+ Integer expected = groundTruth.get(key);
+ if (expected == null) {
+ assertNull(map.get(key));
+ }
+ else {
+ Integer contained = map.get(key);
+ assertNotNull(contained);
+ assertEquals(expected, contained);
+ }
+ }
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testConjunctTraversal() {
+ try {
+ final Random rootRnd = new Random(654685486325439L);
+
+ final int numMaps = 7;
+ final int numKeys = 1000000;
+
+ // ------ create a set of maps ------
+ @SuppressWarnings("unchecked")
+ final KeyMap<Integer, Integer>[] maps = (KeyMap<Integer, Integer>[]) new KeyMap<?, ?>[numMaps];
+ for (int i = 0; i < numMaps; i++) {
+ maps[i] = new KeyMap<>();
+ }
+
+ // ------ prepare probabilities for maps ------
+ final double[] probabilities = new double[numMaps];
+ final double[] probabilitiesTemp = new double[numMaps];
+ {
+ probabilities[0] = 0.5;
+ double remainingProb = 1.0 - probabilities[0];
+ for (int i = 1; i < numMaps - 1; i++) {
+ remainingProb /= 2;
+ probabilities[i] = remainingProb;
+ }
+
+ // compensate for rounding errors
+ probabilities[numMaps - 1] = remainingProb;
+ }
+
+ // ------ generate random elements ------
+ final long probSeed = rootRnd.nextLong();
+ final long keySeed = rootRnd.nextLong();
+
+ final Random probRnd = new Random(probSeed);
+ final Random keyRnd = new Random(keySeed);
+
+ final int maxStride = Integer.MAX_VALUE / numKeys;
+
+ int totalNumElements = 0;
+ int nextKeyValue = 1;
+
+ for (int i = 0; i < numKeys; i++) {
+ int numCopies = (nextKeyValue % 3) + 1;
+ System.arraycopy(probabilities, 0, probabilitiesTemp, 0, numMaps);
+
+ double totalProb = 1.0;
+ for (int copy = 0; copy < numCopies; copy++) {
+ int pos = drawPosProportionally(probabilitiesTemp, totalProb, probRnd);
+ totalProb -= probabilitiesTemp[pos];
+ probabilitiesTemp[pos] = 0.0;
+
+ Integer boxed = nextKeyValue;
+ Integer previous = maps[pos].put(boxed, boxed);
+ assertNull("Test problem - test does not assign unique maps", previous);
+ }
+
+ totalNumElements += numCopies;
+ nextKeyValue += keyRnd.nextInt(maxStride) + 1;
+ }
+
+
+ // check that all maps contain the total number of elements
+ int numContained = 0;
+ for (KeyMap<?, ?> map : maps) {
+ numContained += map.size();
+ }
+ assertEquals(totalNumElements, numContained);
+
+ // ------ check that all elements can be found in the maps ------
+ keyRnd.setSeed(keySeed);
+
+ numContained = 0;
+ nextKeyValue = 1;
+ for (int i = 0; i < numKeys; i++) {
+ int numCopiesExpected = (nextKeyValue % 3) + 1;
+ int numCopiesContained = 0;
+
+ for (KeyMap<Integer, Integer> map : maps) {
+ Integer val = map.get(nextKeyValue);
+ if (val != null) {
+ assertEquals(nextKeyValue, val.intValue());
+ numCopiesContained++;
+ }
+ }
+
+ assertEquals(numCopiesExpected, numCopiesContained);
+ numContained += numCopiesContained;
+
+ nextKeyValue += keyRnd.nextInt(maxStride) + 1;
+ }
+ assertEquals(totalNumElements, numContained);
+
+ // ------ make a traversal over all keys and validate the keys in the traversal ------
+ final int[] keysStartedAndFinished = { 0, 0 };
+ KeyMap.TraversalEvaluator<Integer, Integer> traversal = new KeyMap.TraversalEvaluator<Integer, Integer>() {
+
+ private int key;
+ private int valueCount;
+
+ @Override
+ public void startNewKey(Integer key) {
+ this.key = key;
+ this.valueCount = 0;
+
+ keysStartedAndFinished[0]++;
+ }
+
+ @Override
+ public void nextValue(Integer value) {
+ assertEquals(this.key, value.intValue());
+ this.valueCount++;
+ }
+
+ @Override
+ public void keyDone() {
+ int expected = (key % 3) + 1;
+ if (expected != valueCount) {
+ fail("Wrong count for key " + key + " ; expected=" + expected + " , count=" + valueCount);
+ }
+
+ keysStartedAndFinished[1]++;
+ }
+ };
+
+ KeyMap.traverseMaps(shuffleArray(maps, rootRnd), traversal, 17);
+
+ assertEquals(numKeys, keysStartedAndFinished[0]);
+ assertEquals(numKeys, keysStartedAndFinished[1]);
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testSizeComparator() {
+ try {
+ KeyMap<String, String> map1 = new KeyMap<>(5);
+ KeyMap<String, String> map2 = new KeyMap<>(80);
+
+ assertTrue(map1.getCurrentTableCapacity() < map2.getCurrentTableCapacity());
+
+ assertTrue(KeyMap.CapacityDescendingComparator.INSTANCE.compare(map1, map1) == 0);
+ assertTrue(KeyMap.CapacityDescendingComparator.INSTANCE.compare(map2, map2) == 0);
+ assertTrue(KeyMap.CapacityDescendingComparator.INSTANCE.compare(map1, map2) > 0);
+ assertTrue(KeyMap.CapacityDescendingComparator.INSTANCE.compare(map2, map1) < 0);
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ // ------------------------------------------------------------------------
+
+ private static int drawPosProportionally(double[] array, double totalProbability, Random rnd) {
+ double val = rnd.nextDouble() * totalProbability;
+
+ double accum = 0;
+ for (int i = 0; i < array.length; i++) {
+ accum += array[i];
+ if (val <= accum && array[i] > 0.0) {
+ return i;
+ }
+ }
+
+ // in case of rounding errors
+ return array.length - 1;
+ }
+
+ private static <E> E[] shuffleArray(E[] array, Random rnd) {
+ E[] target = Arrays.copyOf(array, array.length);
+
+ for (int i = target.length - 1; i > 0; i--) {
+ int swapPos = rnd.nextInt(i + 1);
+ E temp = target[i];
+ target[i] = target[swapPos];
+ target[swapPos] = temp;
+ }
+
+ return target;
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/b8f58fab/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/GroupedProcessingTimeWindowExample.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/GroupedProcessingTimeWindowExample.java b/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/GroupedProcessingTimeWindowExample.java
new file mode 100644
index 0000000..6e0724d
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/GroupedProcessingTimeWindowExample.java
@@ -0,0 +1,167 @@
+/*
+ * 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.examples.windowing;
+
+import org.apache.flink.api.common.functions.ReduceFunction;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.TypeInfoParser;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
+import org.apache.flink.streaming.api.functions.windows.KeyedWindowFunction;
+import org.apache.flink.streaming.runtime.operators.windows.AggregatingProcessingTimeWindowOperator;
+import org.apache.flink.util.Collector;
+
+@SuppressWarnings("serial")
+public class GroupedProcessingTimeWindowExample {
+
+ public static void main(String[] args) throws Exception {
+
+ final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setParallelism(4);
+
+ DataStream<Tuple2<Long, Long>> stream = env
+ .addSource(new RichParallelSourceFunction<Tuple2<Long, Long>>() {
+
+ private volatile boolean running = true;
+
+ @Override
+ public void run(SourceContext<Tuple2<Long, Long>> ctx) throws Exception {
+
+ final long startTime = System.currentTimeMillis();
+
+ final long numElements = 20000000;
+ final long numKeys = 10000;
+ long val = 1L;
+ long count = 0L;
+
+
+ while (running && count < numElements) {
+ count++;
+ ctx.collect(new Tuple2<Long, Long>(val++, 1L));
+
+ if (val > numKeys) {
+ val = 1L;
+ }
+ }
+
+ final long endTime = System.currentTimeMillis();
+ System.out.println("Took " + (endTime-startTime) + " msecs for " + numElements + " values");
+ }
+
+ @Override
+ public void cancel() {
+ running = false;
+ }
+ });
+
+ stream
+// .groupBy(new FirstFieldKeyExtractor<Tuple2<Long, Long>, Long>())
+// .window(Time.of(2500, TimeUnit.MILLISECONDS)).every(Time.of(500, TimeUnit.MILLISECONDS))
+// .reduceWindow(new SummingReducer())
+// .flatten()
+// .partitionByHash(new FirstFieldKeyExtractor<Tuple2<Long, Long>, Long>())
+// .transform(
+// "Aligned time window",
+// TypeInfoParser.<Tuple2<Long, Long>>parse("Tuple2<Long, Long>"),
+// new AccumulatingProcessingTimeWindowOperator<Long, Tuple2<Long, Long>, Tuple2<Long, Long>>(
+// new SummingWindowFunction<Long>(),
+// new FirstFieldKeyExtractor<Tuple2<Long, Long>, Long>(),
+// 2500, 500))
+ .transform(
+ "Aligned time window",
+ TypeInfoParser.<Tuple2<Long, Long>>parse("Tuple2<Long, Long>"),
+ new AggregatingProcessingTimeWindowOperator<Long, Tuple2<Long, Long>>(
+ new SummingReducer(),
+ new FirstFieldKeyExtractor<Tuple2<Long, Long>, Long>(),
+ 2500, 500))
+
+ .addSink(new SinkFunction<Tuple2<Long, Long>>() {
+ @Override
+ public void invoke(Tuple2<Long, Long> value) {
+ }
+ });
+
+ env.execute();
+ }
+
+ public static class FirstFieldKeyExtractor<Type extends Tuple, Key> implements KeySelector<Type, Key> {
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public Key getKey(Type value) {
+ return (Key) value.getField(0);
+ }
+ }
+
+ public static class IdentityKeyExtractor<T> implements KeySelector<T, T> {
+
+ @Override
+ public T getKey(T value) {
+ return value;
+ }
+ }
+
+ public static class IdentityWindowFunction<K, T> implements KeyedWindowFunction<K, T, T> {
+
+ @Override
+ public void evaluate(K k, Iterable<T> values, Collector<T> out) throws Exception {
+ for (T v : values) {
+ out.collect(v);
+ }
+ }
+ }
+
+ public static class CountingWindowFunction<K, T> implements KeyedWindowFunction<K, T, Long> {
+
+ @Override
+ public void evaluate(K k, Iterable<T> values, Collector<Long> out) throws Exception {
+ long count = 0;
+ for (T ignored : values) {
+ count++;
+ }
+
+ out.collect(count);
+ }
+ }
+
+ public static class SummingWindowFunction<K> implements KeyedWindowFunction<K, Tuple2<K, Long>, Tuple2<K, Long>> {
+
+ @Override
+ public void evaluate(K key, Iterable<Tuple2<K, Long>> values, Collector<Tuple2<K, Long>> out) throws Exception {
+ long sum = 0L;
+ for (Tuple2<K, Long> value : values) {
+ sum += value.f1;
+ }
+
+ out.collect(new Tuple2<K, Long>(key, sum));
+ }
+ }
+
+ public static class SummingReducer implements ReduceFunction<Tuple2<Long, Long>> {
+
+ @Override
+ public Tuple2<Long, Long> reduce(Tuple2<Long, Long> value1, Tuple2<Long, Long> value2) {
+ return new Tuple2<>(value1.f0, value1.f1 + value2.f1);
+ }
+ }
+}
[3/3] flink git commit: [FLINK-2675] [streaming] Add utilities for
scheduled triggers.
Posted by se...@apache.org.
[FLINK-2675] [streaming] Add utilities for scheduled triggers.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/7ad21031
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/7ad21031
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/7ad21031
Branch: refs/heads/master
Commit: 7ad210316fe22441761351e9943cdc31570d5407
Parents: 64e1dc6
Author: Stephan Ewen <se...@apache.org>
Authored: Tue Sep 15 14:55:41 2015 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Tue Sep 22 13:25:20 2015 +0200
----------------------------------------------------------------------
.../taskmanager/DispatcherThreadFactory.java | 50 +++++++
.../taskmanager/DispatherThreadFactory.java | 50 -------
.../apache/flink/runtime/taskmanager/Task.java | 2 +-
.../runtime/operators/TriggerTimer.java | 119 ++++++++++++++++
.../runtime/operators/Triggerable.java | 38 +++++
.../runtime/operators/package-info.java | 22 +++
.../runtime/operators/TriggerTimerTest.java | 137 +++++++++++++++++++
7 files changed, 367 insertions(+), 51 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad21031/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/DispatcherThreadFactory.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/DispatcherThreadFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/DispatcherThreadFactory.java
new file mode 100644
index 0000000..97060a8
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/DispatcherThreadFactory.java
@@ -0,0 +1,50 @@
+/*
+ * 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.runtime.taskmanager;
+
+import java.util.concurrent.ThreadFactory;
+
+/**
+ * Thread factory that creates threads with a given name, associates them with a given
+ * thread group, and set them to daemon mode.
+ */
+public class DispatcherThreadFactory implements ThreadFactory {
+
+ private final ThreadGroup group;
+
+ private final String threadName;
+
+ /**
+ * Creates a new thread factory.
+ *
+ * @param group The group that the threads will be associated with.
+ * @param threadName The name for the threads.
+ */
+ public DispatcherThreadFactory(ThreadGroup group, String threadName) {
+ this.group = group;
+ this.threadName = threadName;
+ }
+
+ @Override
+ public Thread newThread(Runnable r) {
+ Thread t = new Thread(group, r, threadName);
+ t.setDaemon(true);
+ return t;
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad21031/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/DispatherThreadFactory.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/DispatherThreadFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/DispatherThreadFactory.java
deleted file mode 100644
index f5f1565..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/DispatherThreadFactory.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.flink.runtime.taskmanager;
-
-import java.util.concurrent.ThreadFactory;
-
-/**
- * Thread factory that creates threads with a given name, associates them with a given
- * thread group, and set them to daemon mode.
- */
-public class DispatherThreadFactory implements ThreadFactory {
-
- private final ThreadGroup group;
-
- private final String threadName;
-
- /**
- * Creates a new thread factory.
- *
- * @param group The group that the threads will be associated with.
- * @param threadName The name for the threads.
- */
- public DispatherThreadFactory(ThreadGroup group, String threadName) {
- this.group = group;
- this.threadName = threadName;
- }
-
- @Override
- public Thread newThread(Runnable r) {
- Thread t = new Thread(group, r, threadName);
- t.setDaemon(true);
- return t;
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad21031/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
index 7eff45a..434c5d1 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
@@ -998,7 +998,7 @@ public class Task implements Runnable {
if (executor == null) {
// first time use, initialize
executor = Executors.newSingleThreadExecutor(
- new DispatherThreadFactory(TASK_THREADS_GROUP, "Async calls on " + taskNameWithSubtask));
+ new DispatcherThreadFactory(TASK_THREADS_GROUP, "Async calls on " + taskNameWithSubtask));
this.asyncCallDispatcher = executor;
// double-check for execution state, and make sure we clean up after ourselves
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad21031/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/operators/TriggerTimer.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/operators/TriggerTimer.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/operators/TriggerTimer.java
new file mode 100644
index 0000000..7528eb3
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/operators/TriggerTimer.java
@@ -0,0 +1,119 @@
+/*
+ * 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.runtime.operators;
+
+import org.apache.flink.runtime.taskmanager.DispatcherThreadFactory;
+
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * A timer that triggers targets at a specific point in the future. This timer executes single-threaded,
+ * which means that never more than one trigger will be executed at the same time.
+ * <p>
+ * This timer generally maintains order of trigger events. This means that for two triggers scheduled at
+ * different times, the one scheduled for the later time will be executed after the one scheduled for the
+ * earlier time.
+ */
+public class TriggerTimer {
+
+ /** The thread group that holds all trigger timer threads */
+ public static final ThreadGroup TRIGGER_THREADS_GROUP = new ThreadGroup("Triggers");
+
+ /** The executor service that */
+ private final ScheduledExecutorService scheduler;
+
+
+ /**
+ * Creates a new trigger timer, where the timer thread has the default name "TriggerTimer Thread".
+ */
+ public TriggerTimer() {
+ this("TriggerTimer Thread");
+ }
+
+ /**
+ * Creates a new trigger timer, where the timer thread has the given name.
+ *
+ * @param triggerName The name for the trigger thread.
+ */
+ public TriggerTimer(String triggerName) {
+ this.scheduler = Executors.newSingleThreadScheduledExecutor(
+ new DispatcherThreadFactory(TRIGGER_THREADS_GROUP, triggerName));
+ }
+
+ /**
+ * Schedules a new trigger event. The trigger event will occur roughly at the given timestamp.
+ * If the timestamp is in the past (or now), the trigger will be queued for immediate execution. Note that other
+ * triggers that are to be executed now will be executed before this trigger.
+ *
+ * @param target The target to be triggered.
+ * @param timestamp The timestamp when the trigger should occur, and the timestamp given
+ * to the trigger-able target.
+ */
+ public void scheduleTriggerAt(Triggerable target, long timestamp) {
+ long delay = Math.max(timestamp - System.currentTimeMillis(), 0);
+
+ scheduler.schedule(
+ new TriggerTask(target, timestamp),
+ delay,
+ TimeUnit.MILLISECONDS);
+ }
+
+ /**
+ * Shuts down the trigger timer, canceling all pending triggers and stopping the trigger thread.
+ */
+ public void shutdown() {
+ scheduler.shutdownNow();
+ }
+
+ /**
+ * The finalize method shuts down the timer. This is a fail-safe shutdown, in case the original
+ * shutdown method was never called.
+ * <p>
+ * This should not be relied upon! It will cause shutdown to happen much later than if manual
+ * shutdown is attempted, and cause threads to linger for longer than needed.
+ */
+ @Override
+ @SuppressWarnings("FinalizeDoesntCallSuperFinalize")
+ protected void finalize() {
+ shutdown();
+ }
+
+ // ------------------------------------------------------------------------
+
+ /**
+ * Internal task that is invoked by the scheduled executor and triggers the target.
+ */
+ private static final class TriggerTask implements Runnable {
+
+ private final Triggerable target;
+ private final long timestamp;
+
+ TriggerTask(Triggerable target, long timestamp) {
+ this.target = target;
+ this.timestamp = timestamp;
+ }
+
+ @Override
+ public void run() {
+ target.trigger(timestamp);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad21031/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/operators/Triggerable.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/operators/Triggerable.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/operators/Triggerable.java
new file mode 100644
index 0000000..626f087
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/operators/Triggerable.java
@@ -0,0 +1,38 @@
+/*
+ * 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.runtime.operators;
+
+/**
+ * This interface must be implemented by objects that are triggered by a
+ * {@link TriggerTimer}.
+ */
+public interface Triggerable {
+
+ /**
+ * This method is invoked by the {@link TriggerTimer}
+ * and given the timestamp for which the trigger was scheduled.
+ * <p>
+ * If the triggering is delayed for whatever reason (trigger timer was blocked, JVM stalled due
+ * to a garbage collection), the timestamp supplied to this function will still be the original
+ * timestamp for which the trigger was scheduled.
+ *
+ * @param timestamp The timestamp for which the trigger event was scheduled.
+ */
+ void trigger(long timestamp);
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad21031/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/operators/package-info.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/operators/package-info.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/operators/package-info.java
new file mode 100644
index 0000000..5fe6873
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/operators/package-info.java
@@ -0,0 +1,22 @@
+/*
+ * 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.
+ */
+
+/**
+ * This package contains the operators that perform the stream transformations.
+ * One or more operators are bundled into a "chain" and executed in a stream task.
+ */
+package org.apache.flink.streaming.runtime.operators;
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad21031/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/runtime/operators/TriggerTimerTest.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/runtime/operators/TriggerTimerTest.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/runtime/operators/TriggerTimerTest.java
new file mode 100644
index 0000000..1349cb3
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/runtime/operators/TriggerTimerTest.java
@@ -0,0 +1,137 @@
+/*
+ * 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.runtime.operators;
+
+import org.junit.Test;
+
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.junit.Assert.*;
+
+public class TriggerTimerTest {
+
+ @Test
+ public void testThreadGroupAndShutdown() {
+ try {
+ TriggerTimer timer = new TriggerTimer();
+
+ // first one spawns thread
+ timer.scheduleTriggerAt(new Triggerable() {
+ @Override
+ public void trigger(long timestamp) {}
+ }, System.currentTimeMillis());
+
+ assertEquals(1, TriggerTimer.TRIGGER_THREADS_GROUP.activeCount());
+
+ // thread needs to die in time
+ timer.shutdown();
+
+ long deadline = System.currentTimeMillis() + 2000;
+ while (TriggerTimer.TRIGGER_THREADS_GROUP.activeCount() > 0 && System.currentTimeMillis() < deadline) {
+ Thread.sleep(10);
+ }
+
+ assertEquals("Trigger timer thread did not properly shut down",
+ 0, TriggerTimer.TRIGGER_THREADS_GROUP.activeCount());
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void checkScheduledTimestampe() {
+ try {
+ final TriggerTimer timer = new TriggerTimer();
+
+ final AtomicReference<Throwable> errorRef = new AtomicReference<>();
+
+ final long t1 = System.currentTimeMillis();
+ final long t2 = System.currentTimeMillis() - 200;
+ final long t3 = System.currentTimeMillis() + 100;
+ final long t4 = System.currentTimeMillis() + 200;
+
+ timer.scheduleTriggerAt(new ValidatingTriggerable(errorRef, t1, 0), t1);
+ timer.scheduleTriggerAt(new ValidatingTriggerable(errorRef, t2, 1), t2);
+ timer.scheduleTriggerAt(new ValidatingTriggerable(errorRef, t3, 2), t3);
+ timer.scheduleTriggerAt(new ValidatingTriggerable(errorRef, t4, 3), t4);
+
+ long deadline = System.currentTimeMillis() + 5000;
+ while (errorRef.get() == null &&
+ ValidatingTriggerable.numInSequence < 4 &&
+ System.currentTimeMillis() < deadline)
+ {
+ Thread.sleep(100);
+ }
+
+ // handle errors
+ if (errorRef.get() != null) {
+ errorRef.get().printStackTrace();
+ fail(errorRef.get().getMessage());
+ }
+
+ assertEquals(4, ValidatingTriggerable.numInSequence);
+
+ timer.shutdown();
+
+ // wait until the trigger thread is shut down. otherwise, the other tests may become unstable
+ deadline = System.currentTimeMillis() + 2000;
+ while (TriggerTimer.TRIGGER_THREADS_GROUP.activeCount() > 0 && System.currentTimeMillis() < deadline) {
+ Thread.sleep(10);
+ }
+
+ assertEquals("Trigger timer thread did not properly shut down",
+ 0, TriggerTimer.TRIGGER_THREADS_GROUP.activeCount());
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+
+ private static class ValidatingTriggerable implements Triggerable {
+
+ static int numInSequence;
+
+ private final AtomicReference<Throwable> errorRef;
+
+ private final long expectedTimestamp;
+ private final int expectedInSequence;
+
+ private ValidatingTriggerable(AtomicReference<Throwable> errorRef, long expectedTimestamp, int expectedInSequence) {
+ this.errorRef = errorRef;
+ this.expectedTimestamp = expectedTimestamp;
+ this.expectedInSequence = expectedInSequence;
+ }
+
+ @Override
+ public void trigger(long timestamp) {
+ try {
+ assertEquals(expectedTimestamp, timestamp);
+ assertEquals(expectedInSequence, numInSequence);
+ numInSequence++;
+ }
+ catch (Throwable t) {
+ errorRef.compareAndSet(null, t);
+ }
+ }
+ }
+}
[2/3] flink git commit: [FLINK-2683] [FLINK-2682] [runtime] Add
dedicated operator for aligned processing time windows.
Posted by se...@apache.org.
[FLINK-2683] [FLINK-2682] [runtime] Add dedicated operator for aligned processing time windows.
Also add utilities for heap-backed keyed state in panes (dedicated tailored hash table)
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/b8f58fab
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/b8f58fab
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/b8f58fab
Branch: refs/heads/master
Commit: b8f58fab56c727b8d44d2a90bbb389ab7544989e
Parents: 7ad2103
Author: Stephan Ewen <se...@apache.org>
Authored: Tue Sep 15 15:00:17 2015 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Tue Sep 22 13:25:20 2015 +0200
----------------------------------------------------------------------
.../flink/api/common/functions/Function.java | 2 +-
.../flink/runtime/util/UnionIterator.java | 100 +++
.../flink/runtime/util/UnionIteratorTest.java | 142 ++++
.../functions/windows/KeyedWindowFunction.java | 44 ++
.../api/operators/TimestampedCollector.java | 6 +-
.../operators/StreamingOperatorMetrics.java | 27 +
...ractAlignedProcessingTimeWindowOperator.java | 270 ++++++++
.../windows/AbstractKeyedTimePanes.java | 76 +++
.../windows/AccumulatingKeyedTimePanes.java | 126 ++++
...ccumulatingProcessingTimeWindowOperator.java | 48 ++
.../windows/AggregatingKeyedTimePanes.java | 103 +++
...AggregatingProcessingTimeWindowOperator.java | 47 ++
.../runtime/operators/windows/KeyMap.java | 651 +++++++++++++++++++
.../runtime/operators/windows/package-info.java | 22 +
.../runtime/tasks/StreamTaskException.java | 2 +-
.../streaming/runtime/tasks/package-info.java | 27 +
...AlignedProcessingTimeWindowOperatorTest.java | 515 +++++++++++++++
...AlignedProcessingTimeWindowOperatorTest.java | 528 +++++++++++++++
.../operators/windows/CollectingOutput.java | 80 +++
.../windows/KeyMapPutIfAbsentTest.java | 121 ++++
.../operators/windows/KeyMapPutTest.java | 136 ++++
.../runtime/operators/windows/KeyMapTest.java | 344 ++++++++++
.../GroupedProcessingTimeWindowExample.java | 167 +++++
23 files changed, 3581 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/b8f58fab/flink-core/src/main/java/org/apache/flink/api/common/functions/Function.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/functions/Function.java b/flink-core/src/main/java/org/apache/flink/api/common/functions/Function.java
index 632a0d2..8a1819c 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/functions/Function.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/functions/Function.java
@@ -24,5 +24,5 @@ package org.apache.flink.api.common.functions;
* <p>This interface is empty in order to allow extending interfaces to
* be SAM (single abstract method) interfaces that can be implemented via Java 8 lambdas.</p>
*/
-public interface Function {
+public interface Function extends java.io.Serializable {
}
http://git-wip-us.apache.org/repos/asf/flink/blob/b8f58fab/flink-runtime/src/main/java/org/apache/flink/runtime/util/UnionIterator.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/util/UnionIterator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/util/UnionIterator.java
new file mode 100644
index 0000000..c279adf
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/util/UnionIterator.java
@@ -0,0 +1,100 @@
+/*
+ * 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.runtime.util;
+
+import org.apache.flink.util.TraversableOnceException;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.NoSuchElementException;
+
+public class UnionIterator<T> implements Iterator<T>, Iterable<T> {
+
+ private Iterator<T> currentIterator;
+
+ private ArrayList<List<T>> furtherLists = new ArrayList<>();
+
+ private int nextList;
+
+ private boolean iteratorAvailable = true;
+
+ // ------------------------------------------------------------------------
+
+ public void clear() {
+ currentIterator = null;
+ furtherLists.clear();
+ nextList = 0;
+ iteratorAvailable = true;
+ }
+
+ public void addList(List<T> list) {
+ if (currentIterator == null) {
+ currentIterator = list.iterator();
+ }
+ else {
+ furtherLists.add(list);
+ }
+ }
+
+ // ------------------------------------------------------------------------
+
+ @Override
+ public Iterator<T> iterator() {
+ if (iteratorAvailable) {
+ iteratorAvailable = false;
+ return this;
+ } else {
+ throw new TraversableOnceException();
+ }
+ }
+
+ @Override
+ public boolean hasNext() {
+ while (currentIterator != null) {
+ if (currentIterator.hasNext()) {
+ return true;
+ }
+ else if (nextList < furtherLists.size()) {
+ currentIterator = furtherLists.get(nextList).iterator();
+ nextList++;
+ }
+ else {
+ currentIterator = null;
+ }
+ }
+
+ return false;
+ }
+
+ @Override
+ public T next() {
+ if (hasNext()) {
+ return currentIterator.next();
+ }
+ else {
+ throw new NoSuchElementException();
+ }
+ }
+
+ @Override
+ public void remove() {
+ throw new UnsupportedOperationException();
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/b8f58fab/flink-runtime/src/test/java/org/apache/flink/runtime/util/UnionIteratorTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/util/UnionIteratorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/util/UnionIteratorTest.java
new file mode 100644
index 0000000..2a79b1f
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/util/UnionIteratorTest.java
@@ -0,0 +1,142 @@
+/*
+ * 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.runtime.util;
+
+import org.apache.flink.util.TraversableOnceException;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.NoSuchElementException;
+
+import static org.junit.Assert.*;
+
+public class UnionIteratorTest {
+
+ @Test
+ public void testUnion() {
+ try {
+ UnionIterator<Integer> iter = new UnionIterator<>();
+
+ // should succeed and be empty
+ assertFalse(iter.iterator().hasNext());
+
+ iter.clear();
+
+ try {
+ iter.iterator().next();
+ fail("should fail with an exception");
+ } catch (NoSuchElementException e) {
+ // expected
+ }
+
+ iter.clear();
+ iter.addList(Arrays.asList(1, 2, 3, 4, 5, 6, 7));
+ iter.addList(Collections.<Integer>emptyList());
+ iter.addList(Arrays.asList(8, 9, 10, 11));
+
+ int val = 1;
+ for (int i : iter) {
+ assertEquals(val++, i);
+ }
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testTraversableOnce() {
+ try {
+ UnionIterator<Integer> iter = new UnionIterator<>();
+
+ // should succeed
+ iter.iterator();
+
+ // should fail
+ try {
+ iter.iterator();
+ fail("should fail with an exception");
+ } catch (TraversableOnceException e) {
+ // expected
+ }
+
+ // should fail again
+ try {
+ iter.iterator();
+ fail("should fail with an exception");
+ } catch (TraversableOnceException e) {
+ // expected
+ }
+
+ // reset the thing, keep it empty
+ iter.clear();
+
+ // should succeed
+ iter.iterator();
+
+ // should fail
+ try {
+ iter.iterator();
+ fail("should fail with an exception");
+ } catch (TraversableOnceException e) {
+ // expected
+ }
+
+ // should fail again
+ try {
+ iter.iterator();
+ fail("should fail with an exception");
+ } catch (TraversableOnceException e) {
+ // expected
+ }
+
+ // reset the thing, add some data
+ iter.clear();
+ iter.addList(Arrays.asList(1, 2, 3, 4, 5, 6, 7));
+
+ // should succeed
+ Iterator<Integer> ints = iter.iterator();
+ assertNotNull(ints.next());
+ assertNotNull(ints.next());
+ assertNotNull(ints.next());
+
+ // should fail if called in the middle of operations
+ try {
+ iter.iterator();
+ fail("should fail with an exception");
+ } catch (TraversableOnceException e) {
+ // expected
+ }
+
+ // reset the thing, keep it empty
+ iter.clear();
+
+ // should succeed again
+ assertFalse(iter.iterator().hasNext());
+
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/b8f58fab/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/windows/KeyedWindowFunction.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/windows/KeyedWindowFunction.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/windows/KeyedWindowFunction.java
new file mode 100644
index 0000000..d7ca0a1
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/windows/KeyedWindowFunction.java
@@ -0,0 +1,44 @@
+/*
+ * 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.functions.windows;
+
+import org.apache.flink.api.common.functions.Function;
+import org.apache.flink.util.Collector;
+
+import java.io.Serializable;
+
+/**
+ * Base interface for functions that are evaluated over keyed (grouped) windows.
+ *
+ * @param <KEY> The type of the key.
+ * @param <IN> The type of the input value.
+ * @param <OUT> The type of the output value.
+ */
+public interface KeyedWindowFunction<KEY, IN, OUT> extends Function, Serializable {
+
+ /**
+ *
+ * @param key
+ * @param values
+ * @param out
+ *
+ * @throws Exception The function may throw exceptions to fail the program and trigger recovery.
+ */
+ void evaluate(KEY key, Iterable<IN> values, Collector<OUT> out) throws Exception;
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/b8f58fab/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/TimestampedCollector.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/TimestampedCollector.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/TimestampedCollector.java
index 0ff223c..62514fc 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/TimestampedCollector.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/TimestampedCollector.java
@@ -30,9 +30,13 @@ import org.joda.time.Instant;
* @param <T> The type of the elments that can be emitted.
*/
public class TimestampedCollector<T> implements Collector<T> {
+
private final Output<StreamRecord<T>> output;
+
+ private final StreamRecord<T> reuse;
+
private long timestamp;
- private StreamRecord<T> reuse;
+
/**
* Creates a new {@link TimestampedCollector} that wraps the given {@link Output}.
http://git-wip-us.apache.org/repos/asf/flink/blob/b8f58fab/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/operators/StreamingOperatorMetrics.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/operators/StreamingOperatorMetrics.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/operators/StreamingOperatorMetrics.java
new file mode 100644
index 0000000..8429889
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/operators/StreamingOperatorMetrics.java
@@ -0,0 +1,27 @@
+/*
+ * 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.runtime.operators;
+
+public class StreamingOperatorMetrics {
+
+
+ public void incrementLateElementDiscarded() {
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/b8f58fab/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/operators/windows/AbstractAlignedProcessingTimeWindowOperator.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/operators/windows/AbstractAlignedProcessingTimeWindowOperator.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/operators/windows/AbstractAlignedProcessingTimeWindowOperator.java
new file mode 100644
index 0000000..f5f576d
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/operators/windows/AbstractAlignedProcessingTimeWindowOperator.java
@@ -0,0 +1,270 @@
+/*
+ * 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.runtime.operators.windows;
+
+import org.apache.commons.math3.util.ArithmeticUtils;
+
+import org.apache.flink.api.common.functions.Function;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.util.MathUtils;
+import org.apache.flink.runtime.util.SerializableObject;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.operators.TimestampedCollector;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.operators.TriggerTimer;
+import org.apache.flink.streaming.runtime.operators.Triggerable;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+
+public abstract class AbstractAlignedProcessingTimeWindowOperator<KEY, IN, OUT> extends AbstractStreamOperator<OUT>
+ implements OneInputStreamOperator<IN, OUT>, Triggerable {
+
+ private static final long serialVersionUID = 3245500864882459867L;
+
+ private static final long MIN_SLIDE_TIME = 50;
+
+
+ private final SerializableObject lock = new SerializableObject();
+
+ // ----- fields for operator parametrization -----
+
+ private final Function function;
+ private final KeySelector<IN, KEY> keySelector;
+
+ private final long windowSize;
+ private final long windowSlide;
+ private final long paneSize;
+ private final int numPanesPerWindow;
+
+ // ----- fields for operator functionality -----
+
+ private transient AbstractKeyedTimePanes<IN, KEY, ?, OUT> panes;
+
+ private transient TimestampedCollector<OUT> out;
+
+ private transient TriggerTimer triggerTimer;
+
+ private transient long nextEvaluationTime;
+ private transient long nextSlideTime;
+
+ private transient volatile Throwable asyncError;
+
+
+ protected AbstractAlignedProcessingTimeWindowOperator(
+ Function function,
+ KeySelector<IN, KEY> keySelector,
+ long windowLength,
+ long windowSlide)
+ {
+ if (function == null || keySelector == null) {
+ throw new NullPointerException();
+ }
+ if (windowLength < MIN_SLIDE_TIME) {
+ throw new IllegalArgumentException("Window length must be at least " + MIN_SLIDE_TIME + " msecs");
+ }
+ if (windowSlide < MIN_SLIDE_TIME) {
+ throw new IllegalArgumentException("Window slide must be at least " + MIN_SLIDE_TIME + " msecs");
+ }
+ if (windowLength < windowSlide) {
+ throw new IllegalArgumentException("The window size must be larger than the window slide");
+ }
+
+ final long paneSlide = ArithmeticUtils.gcd(windowLength, windowSlide);
+ if (paneSlide < MIN_SLIDE_TIME) {
+ throw new IllegalArgumentException(String.format(
+ "Cannot compute window of size %d msecs sliding by %d msecs. " +
+ "The unit of grouping is too small: %d msecs", windowLength, windowSlide, paneSlide));
+ }
+
+ this.function = function;
+ this.keySelector = keySelector;
+ this.windowSize = windowLength;
+ this.windowSlide = windowSlide;
+ this.paneSize = paneSlide;
+ this.numPanesPerWindow = MathUtils.checkedDownCast(windowLength / paneSlide);
+ }
+
+
+ protected abstract AbstractKeyedTimePanes<IN, KEY, ?, OUT> createPanes(
+ KeySelector<IN, KEY> keySelector, Function function);
+
+ // ------------------------------------------------------------------------
+ // startup and shutdown
+ // ------------------------------------------------------------------------
+
+ @Override
+ public void open(Configuration parameters) throws Exception {
+ out = new TimestampedCollector<>(output);
+
+ // create the panes that gather the elements per slide
+ panes = createPanes(keySelector, function);
+
+ // decide when to first compute the window and when to slide it
+ // the values should align with the start of time (that is, the UNIX epoch, not the big bang)
+ final long now = System.currentTimeMillis();
+ nextEvaluationTime = now + windowSlide - (now % windowSlide);
+ nextSlideTime = now + paneSize - (now % paneSize);
+
+ // start the trigger timer
+ triggerTimer = new TriggerTimer("Trigger for " + getRuntimeContext().getTaskName());
+
+ // schedule the first trigger
+ triggerTimer.scheduleTriggerAt(this, Math.min(nextEvaluationTime, nextSlideTime));
+ }
+
+ @Override
+ public void close() throws Exception {
+ // acquire the lock during shutdown, to prevent trigger calls at the same time
+ synchronized (lock) {
+ final long finalWindowTimestamp = nextEvaluationTime;
+
+ // early stop the triggering thread, so it does not attempt to return any more data
+ stopTriggers();
+
+ // make sure we had no asynchronous error so far
+ checkErroneous();
+
+ // emit the remaining data
+ computeWindow(finalWindowTimestamp);
+ }
+ }
+
+ @Override
+ public void dispose() {
+ // acquire the lock during shutdown, to prevent trigger calls at the same time
+ synchronized (lock) {
+ // fail-safe stop of the triggering thread (in case of an error)
+ stopTriggers();
+
+ // release the panes
+ panes.dispose();
+ }
+ }
+
+ private void stopTriggers() {
+ if (triggerTimer != null) {
+ triggerTimer.shutdown();
+ }
+
+ // reset the action timestamps. this makes sure any pending triggers will not evaluate
+ nextEvaluationTime = -1L;
+ nextSlideTime = -1L;
+ }
+
+ // ------------------------------------------------------------------------
+ // Receiving elements and triggers
+ // ------------------------------------------------------------------------
+
+ @Override
+ public void processElement(StreamRecord<IN> element) throws Exception {
+ synchronized (lock) {
+ checkErroneous();
+ panes.addElementToLatestPane(element.getValue());
+ }
+ }
+
+ @Override
+ public void processWatermark(Watermark mark) {
+ // this operator does not react to watermarks
+ }
+
+ @Override
+ public void trigger(long timestamp) {
+ synchronized (lock) {
+ // first we check if we actually trigger the window function
+ if (timestamp == nextEvaluationTime) {
+ // compute and output the results
+ try {
+ computeWindow(timestamp);
+ }
+ catch (Throwable t) {
+ this.asyncError = t;
+ }
+
+ nextEvaluationTime += windowSlide;
+ }
+
+ // check if we slide the panes by one. this may happen in addition to the
+ // window computation, or just by itself
+ if (timestamp == nextSlideTime) {
+ try {
+ panes.slidePanes(numPanesPerWindow);
+ }
+ catch (Throwable t) {
+ this.asyncError = t;
+ }
+ nextSlideTime += paneSize;
+ }
+
+ long nextTriggerTime = Math.min(nextEvaluationTime, nextSlideTime);
+ triggerTimer.scheduleTriggerAt(this, nextTriggerTime);
+ }
+ }
+
+ private void checkErroneous() throws Exception {
+ if (asyncError != null) {
+ throw new Exception("Error while computing and producing window result", asyncError);
+ }
+ }
+
+ private void computeWindow(long timestamp) throws Exception {
+ out.setTimestamp(timestamp);
+ panes.truncatePanes(numPanesPerWindow);
+ panes.evaluateWindow(out);
+ }
+
+ // ------------------------------------------------------------------------
+ // Property access (for testing)
+ // ------------------------------------------------------------------------
+
+ public long getWindowSize() {
+ return windowSize;
+ }
+
+ public long getWindowSlide() {
+ return windowSlide;
+ }
+
+ public long getPaneSize() {
+ return paneSize;
+ }
+
+ public int getNumPanesPerWindow() {
+ return numPanesPerWindow;
+ }
+
+ public long getNextEvaluationTime() {
+ return nextEvaluationTime;
+ }
+
+ public long getNextSlideTime() {
+ return nextSlideTime;
+ }
+
+ // ------------------------------------------------------------------------
+ // Utilities
+ // ------------------------------------------------------------------------
+
+ @Override
+ public String toString() {
+ return "Window (processing time) (length=" + windowSize + ", slide=" + windowSlide + ')';
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/b8f58fab/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/operators/windows/AbstractKeyedTimePanes.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/operators/windows/AbstractKeyedTimePanes.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/operators/windows/AbstractKeyedTimePanes.java
new file mode 100644
index 0000000..a49b2e6
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/operators/windows/AbstractKeyedTimePanes.java
@@ -0,0 +1,76 @@
+/*
+ * 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.runtime.operators.windows;
+
+import org.apache.flink.util.Collector;
+
+import java.util.ArrayDeque;
+
+
+public abstract class AbstractKeyedTimePanes<Type, Key, Aggregate, Result> {
+
+ protected KeyMap<Key, Aggregate> latestPane = new KeyMap<>();
+
+ protected final ArrayDeque<KeyMap<Key, Aggregate>> previousPanes = new ArrayDeque<>();
+
+ // ------------------------------------------------------------------------
+
+ public abstract void addElementToLatestPane(Type element) throws Exception;
+
+ public abstract void evaluateWindow(Collector<Result> out) throws Exception;
+
+
+ public void dispose() {
+ // since all is heap data, there is no need to clean up anything
+ latestPane = null;
+ previousPanes.clear();
+ }
+
+
+ public void slidePanes(int panesToKeep) {
+ if (panesToKeep > 1) {
+ // the current pane becomes the latest previous pane
+ previousPanes.addLast(latestPane);
+
+ // truncate the history
+ while (previousPanes.size() >= panesToKeep) {
+ previousPanes.removeFirst();
+ }
+ }
+
+ // we need a new latest pane
+ latestPane = new KeyMap<>();
+ }
+
+ public void truncatePanes(int numToRetain) {
+ while (previousPanes.size() >= numToRetain) {
+ previousPanes.removeFirst();
+ }
+ }
+
+ protected void traverseAllPanes(KeyMap.TraversalEvaluator<Key, Aggregate> traversal, long traversalPass) throws Exception{
+ // gather all panes in an array (faster iterations)
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ KeyMap<Key, Aggregate>[] panes = previousPanes.toArray(new KeyMap[previousPanes.size() + 1]);
+ panes[panes.length - 1] = latestPane;
+
+ // let the maps make a coordinated traversal and evaluate the window function per contained key
+ KeyMap.traverseMaps(panes, traversal, traversalPass);
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/b8f58fab/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/operators/windows/AccumulatingKeyedTimePanes.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/operators/windows/AccumulatingKeyedTimePanes.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/operators/windows/AccumulatingKeyedTimePanes.java
new file mode 100644
index 0000000..e776106
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/operators/windows/AccumulatingKeyedTimePanes.java
@@ -0,0 +1,126 @@
+/*
+ * 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.runtime.operators.windows;
+
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.runtime.util.UnionIterator;
+import org.apache.flink.streaming.api.functions.windows.KeyedWindowFunction;
+import org.apache.flink.util.Collector;
+
+import java.util.ArrayList;
+
+
+public class AccumulatingKeyedTimePanes<Type, Key, Result> extends AbstractKeyedTimePanes<Type, Key, ArrayList<Type>, Result> {
+
+ private final KeySelector<Type, Key> keySelector;
+
+ private final KeyMap.LazyFactory<ArrayList<Type>> listFactory = getListFactory();
+
+ private final KeyedWindowFunction<Key, Type, Result> function;
+
+ private long evaluationPass;
+
+ // ------------------------------------------------------------------------
+
+ public AccumulatingKeyedTimePanes(KeySelector<Type, Key> keySelector, KeyedWindowFunction<Key, Type, Result> function) {
+ this.keySelector = keySelector;
+ this.function = function;
+ }
+
+ // ------------------------------------------------------------------------
+
+ @Override
+ public void addElementToLatestPane(Type element) throws Exception {
+ Key k = keySelector.getKey(element);
+ ArrayList<Type> elements = latestPane.putIfAbsent(k, listFactory);
+ elements.add(element);
+ }
+
+ @Override
+ public void evaluateWindow(Collector<Result> out) throws Exception {
+ if (previousPanes.isEmpty()) {
+ // optimized path for single pane case (tumbling window)
+ for (KeyMap.Entry<Key, ArrayList<Type>> entry : latestPane) {
+ function.evaluate(entry.getKey(), entry.getValue(), out);
+ }
+ }
+ else {
+ // general code path for multi-pane case
+ WindowFunctionTraversal<Key, Type, Result> evaluator = new WindowFunctionTraversal<>(function, out);
+ traverseAllPanes(evaluator, evaluationPass);
+ }
+
+ evaluationPass++;
+ }
+
+ // ------------------------------------------------------------------------
+ // Running a window function in a map traversal
+ // ------------------------------------------------------------------------
+
+ static final class WindowFunctionTraversal<Key, Type, Result> implements KeyMap.TraversalEvaluator<Key, ArrayList<Type>> {
+
+ private final KeyedWindowFunction<Key, Type, Result> function;
+
+ private final UnionIterator<Type> unionIterator;
+
+ private final Collector<Result> out;
+
+ private Key currentKey;
+
+ WindowFunctionTraversal(KeyedWindowFunction<Key, Type, Result> function, Collector<Result> out) {
+ this.function = function;
+ this.out = out;
+ this.unionIterator = new UnionIterator<>();
+ }
+
+
+ @Override
+ public void startNewKey(Key key) {
+ unionIterator.clear();
+ currentKey = key;
+ }
+
+ @Override
+ public void nextValue(ArrayList<Type> value) {
+ unionIterator.addList(value);
+ }
+
+ @Override
+ public void keyDone() throws Exception {
+ function.evaluate(currentKey, unionIterator, out);
+ }
+ }
+
+ // ------------------------------------------------------------------------
+ // Lazy factory for lists (put if absent)
+ // ------------------------------------------------------------------------
+
+ @SuppressWarnings("unchecked")
+ private static <V> KeyMap.LazyFactory<ArrayList<V>> getListFactory() {
+ return (KeyMap.LazyFactory<ArrayList<V>>) LIST_FACTORY;
+ }
+
+ private static final KeyMap.LazyFactory<?> LIST_FACTORY = new KeyMap.LazyFactory<ArrayList<?>>() {
+
+ @Override
+ public ArrayList<?> create() {
+ return new ArrayList<>(4);
+ }
+ };
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/b8f58fab/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/operators/windows/AccumulatingProcessingTimeWindowOperator.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/operators/windows/AccumulatingProcessingTimeWindowOperator.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/operators/windows/AccumulatingProcessingTimeWindowOperator.java
new file mode 100644
index 0000000..16444c2
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/operators/windows/AccumulatingProcessingTimeWindowOperator.java
@@ -0,0 +1,48 @@
+/*
+ * 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.runtime.operators.windows;
+
+import org.apache.flink.api.common.functions.Function;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.streaming.api.functions.windows.KeyedWindowFunction;
+
+
+public class AccumulatingProcessingTimeWindowOperator<KEY, IN, OUT>
+ extends AbstractAlignedProcessingTimeWindowOperator<KEY, IN, OUT> {
+
+ private static final long serialVersionUID = 7305948082830843475L;
+
+
+ public AccumulatingProcessingTimeWindowOperator(
+ KeyedWindowFunction<KEY, IN, OUT> function,
+ KeySelector<IN, KEY> keySelector,
+ long windowLength,
+ long windowSlide)
+ {
+ super(function, keySelector, windowLength, windowSlide);
+ }
+
+ @Override
+ protected AccumulatingKeyedTimePanes<IN, KEY, OUT> createPanes(KeySelector<IN, KEY> keySelector, Function function) {
+ @SuppressWarnings("unchecked")
+ KeyedWindowFunction<KEY, IN, OUT> windowFunction = (KeyedWindowFunction<KEY, IN, OUT>) function;
+
+ return new AccumulatingKeyedTimePanes<>(keySelector, windowFunction);
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/b8f58fab/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/operators/windows/AggregatingKeyedTimePanes.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/operators/windows/AggregatingKeyedTimePanes.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/operators/windows/AggregatingKeyedTimePanes.java
new file mode 100644
index 0000000..730c984
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/operators/windows/AggregatingKeyedTimePanes.java
@@ -0,0 +1,103 @@
+/*
+ * 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.runtime.operators.windows;
+
+import org.apache.flink.api.common.functions.ReduceFunction;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.util.Collector;
+
+
+public class AggregatingKeyedTimePanes<Type, Key> extends AbstractKeyedTimePanes<Type, Key, Type, Type> {
+
+ private final KeySelector<Type, Key> keySelector;
+
+ private final ReduceFunction<Type> reducer;
+
+ private long evaluationPass;
+
+ // ------------------------------------------------------------------------
+
+ public AggregatingKeyedTimePanes(KeySelector<Type, Key> keySelector, ReduceFunction<Type> reducer) {
+ this.keySelector = keySelector;
+ this.reducer = reducer;
+ }
+
+ // ------------------------------------------------------------------------
+
+ @Override
+ public void addElementToLatestPane(Type element) throws Exception {
+ Key k = keySelector.getKey(element);
+ latestPane.putOrAggregate(k, element, reducer);
+ }
+
+ @Override
+ public void evaluateWindow(Collector<Type> out) throws Exception {
+ if (previousPanes.isEmpty()) {
+ // optimized path for single pane case
+ for (KeyMap.Entry<Key, Type> entry : latestPane) {
+ out.collect(entry.getValue());
+ }
+ }
+ else {
+ // general code path for multi-pane case
+ AggregatingTraversal<Key, Type> evaluator = new AggregatingTraversal<>(reducer, out);
+ traverseAllPanes(evaluator, evaluationPass);
+ }
+
+ evaluationPass++;
+ }
+
+ // ------------------------------------------------------------------------
+ // The maps traversal that performs the final aggregation
+ // ------------------------------------------------------------------------
+
+ static final class AggregatingTraversal<Key, Type> implements KeyMap.TraversalEvaluator<Key, Type> {
+
+ private final ReduceFunction<Type> function;
+
+ private final Collector<Type> out;
+
+ private Type currentValue;
+
+ AggregatingTraversal(ReduceFunction<Type> function, Collector<Type> out) {
+ this.function = function;
+ this.out = out;
+ }
+
+ @Override
+ public void startNewKey(Key key) {
+ currentValue = null;
+ }
+
+ @Override
+ public void nextValue(Type value) throws Exception {
+ if (currentValue != null) {
+ currentValue = function.reduce(currentValue, value);
+ }
+ else {
+ currentValue = value;
+ }
+ }
+
+ @Override
+ public void keyDone() throws Exception {
+ out.collect(currentValue);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/b8f58fab/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/operators/windows/AggregatingProcessingTimeWindowOperator.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/operators/windows/AggregatingProcessingTimeWindowOperator.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/operators/windows/AggregatingProcessingTimeWindowOperator.java
new file mode 100644
index 0000000..8bed749
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/operators/windows/AggregatingProcessingTimeWindowOperator.java
@@ -0,0 +1,47 @@
+/*
+ * 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.runtime.operators.windows;
+
+import org.apache.flink.api.common.functions.Function;
+import org.apache.flink.api.common.functions.ReduceFunction;
+import org.apache.flink.api.java.functions.KeySelector;
+
+public class AggregatingProcessingTimeWindowOperator<KEY, IN>
+ extends AbstractAlignedProcessingTimeWindowOperator<KEY, IN, IN> {
+
+ private static final long serialVersionUID = 7305948082830843475L;
+
+
+ public AggregatingProcessingTimeWindowOperator(
+ ReduceFunction<IN> function,
+ KeySelector<IN, KEY> keySelector,
+ long windowLength,
+ long windowSlide)
+ {
+ super(function, keySelector, windowLength, windowSlide);
+ }
+
+ @Override
+ protected AggregatingKeyedTimePanes<IN, KEY> createPanes(KeySelector<IN, KEY> keySelector, Function function) {
+ @SuppressWarnings("unchecked")
+ ReduceFunction<IN> windowFunction = (ReduceFunction<IN>) function;
+
+ return new AggregatingKeyedTimePanes<IN, KEY>(keySelector, windowFunction);
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/b8f58fab/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/operators/windows/KeyMap.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/operators/windows/KeyMap.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/operators/windows/KeyMap.java
new file mode 100644
index 0000000..6e2d75e
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/operators/windows/KeyMap.java
@@ -0,0 +1,651 @@
+/*
+ * 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.runtime.operators.windows;
+
+import org.apache.flink.api.common.functions.ReduceFunction;
+import org.apache.flink.runtime.util.MathUtils;
+
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.Iterator;
+import java.util.NoSuchElementException;
+
+/**
+ * A special Hash Map implementation that can be traversed efficiently in sync with other
+ * hash maps.
+ * <p>
+ * The differences between this hash map and Java's "java.util.HashMap" are:
+ * <ul>
+ * <li>A different hashing scheme. This implementation uses extensible hashing, meaning that
+ * each hash table growth takes one more lower hash code bit into account, and values that where
+ * formerly in the same bucket will afterwards be in the two adjacent buckets.</li>
+ * <li>This allows an efficient traversal of multiple hash maps together, even though the maps are
+ * of different sizes.</li>
+ * <li>The map offers functions such as "putIfAbsent()" and "putOrAggregate()"</li>
+ * <li>The map supports no removal/shrinking.</li>
+ * </ul>
+ */
+public class KeyMap<K, V> implements Iterable<KeyMap.Entry<K, V>> {
+
+ /** The minimum table capacity, 64 entries */
+ private static final int MIN_CAPACITY = 0x40;
+
+ /** The maximum possible table capacity, the largest positive power of
+ * two in the 32bit signed integer value range */
+ private static final int MAX_CAPACITY = 0x40000000;
+
+ /** The number of bits used for table addressing when table is at max capacity */
+ private static final int FULL_BIT_RANGE = MathUtils.log2strict(MAX_CAPACITY);
+
+ // ------------------------------------------------------------------------
+
+ /** The hash index, as an array of entries */
+ private Entry<K, V>[] table;
+
+ /** The number of bits by which the hash code is shifted right, to find the bucket */
+ private int shift;
+
+ /** The number of elements in the hash table */
+ private int numElements;
+
+ /** The number of elements above which the hash table needs to grow */
+ private int rehashThreshold;
+
+ /** The base-2 logarithm of the table capacity */
+ private int log2size;
+
+ // ------------------------------------------------------------------------
+
+ /**
+ * Creates a new hash table with the default initial capacity.
+ */
+ public KeyMap() {
+ this(0);
+ }
+
+ /**
+ * Creates a new table with a capacity tailored to the given expected number of elements.
+ *
+ * @param expectedNumberOfElements The number of elements to tailor the capacity to.
+ */
+ public KeyMap(int expectedNumberOfElements) {
+ if (expectedNumberOfElements < 0) {
+ throw new IllegalArgumentException("Invalid capacity: " + expectedNumberOfElements);
+ }
+
+ // round up to the next power or two
+ // guard against too small capacity and integer overflows
+ int capacity = Integer.highestOneBit(expectedNumberOfElements) << 1;
+ capacity = capacity >= 0 ? Math.max(MIN_CAPACITY, capacity) : MAX_CAPACITY;
+
+ // this also acts as a sanity check
+ log2size = MathUtils.log2strict(capacity);
+ shift = FULL_BIT_RANGE - log2size;
+ table = allocateTable(capacity);
+ rehashThreshold = getRehashThreshold(capacity);
+ }
+
+ // ------------------------------------------------------------------------
+ // Gets and Puts
+ // ------------------------------------------------------------------------
+
+ /**
+ * Inserts the given value, mapped under the given key. If the table already contains a value for
+ * the key, the value is replaced and returned. If no value is contained, yet, the function
+ * returns null.
+ *
+ * @param key The key to insert.
+ * @param value The value to insert.
+ * @return The previously mapped value for the key, or null, if no value was mapped for the key.
+ *
+ * @throws java.lang.NullPointerException Thrown, if the key is null.
+ */
+ public final V put(K key, V value) {
+ final int hash = hash(key);
+ final int slot = indexOf (hash);
+
+ // search the chain from the slot
+ for (Entry<K, V> e = table[slot]; e != null; e = e.next) {
+ Object k;
+ if (e.hashCode == hash && ((k = e.key) == key || key.equals(k))) {
+ // found match
+ V old = e.value;
+ e.value = value;
+ return old;
+ }
+ }
+
+ // no match, insert a new value
+ insertNewEntry(hash, key, value, slot);
+ return null;
+ }
+
+ /**
+ * Inserts a value for the given key, if no value is yet contained for that key. Otherwise,
+ * returns the value currently contained for the key.
+ * <p>
+ * The value that is inserted in case that the key is not contained, yet, is lazily created
+ * using the given factory.
+ *
+ * @param key The key to insert.
+ * @param factory The factory that produces the value, if no value is contained, yet, for the key.
+ * @return The value in the map after this operation (either the previously contained value, or the
+ * newly created value).
+ *
+ * @throws java.lang.NullPointerException Thrown, if the key is null.
+ */
+ public final V putIfAbsent(K key, LazyFactory<V> factory) {
+ final int hash = hash(key);
+ final int slot = indexOf(hash);
+
+ // search the chain from the slot
+ for (Entry<K, V> entry = table[slot]; entry != null; entry = entry.next) {
+ if (entry.hashCode == hash && entry.key.equals(key)) {
+ // found match
+ return entry.value;
+ }
+ }
+
+ // no match, insert a new value
+ V value = factory.create();
+ insertNewEntry(hash, key, value, slot);
+
+ // return the created value
+ return value;
+ }
+
+ /**
+ * Inserts or aggregates a value into the hash map. If the hash map does not yet contain the key,
+ * this method inserts the value. If the table already contains the key (and a value) this
+ * method will use the given ReduceFunction function to combine the existing value and the
+ * given value to a new value, and store that value for the key.
+ *
+ * @param key The key to map the value.
+ * @param value The new value to insert, or aggregate with the existing value.
+ * @param aggregator The aggregator to use if a value is already contained.
+ *
+ * @return The value in the map after this operation: Either the given value, or the aggregated value.
+ *
+ * @throws java.lang.NullPointerException Thrown, if the key is null.
+ * @throws Exception The method forwards exceptions from the aggregation function.
+ */
+ public final V putOrAggregate(K key, V value, ReduceFunction<V> aggregator) throws Exception {
+ final int hash = hash(key);
+ final int slot = indexOf(hash);
+
+ // search the chain from the slot
+ for (Entry<K, V> entry = table[slot]; entry != null; entry = entry.next) {
+ if (entry.hashCode == hash && entry.key.equals(key)) {
+ // found match
+ entry.value = aggregator.reduce(entry.value, value);
+ return entry.value;
+ }
+ }
+
+ // no match, insert a new value
+ insertNewEntry(hash, key, value, slot);
+ // return the original value
+ return value;
+ }
+
+ /**
+ * Looks up the value mapped under the given key. Returns null if no value is mapped under this key.
+ *
+ * @param key The key to look up.
+ * @return The value associated with the key, or null, if no value is found for the key.
+ *
+ * @throws java.lang.NullPointerException Thrown, if the key is null.
+ */
+ public V get(K key) {
+ final int hash = hash(key);
+ final int slot = indexOf(hash);
+
+ // search the chain from the slot
+ for (Entry<K, V> entry = table[slot]; entry != null; entry = entry.next) {
+ if (entry.hashCode == hash && entry.key.equals(key)) {
+ return entry.value;
+ }
+ }
+
+ // not found
+ return null;
+ }
+
+ private void insertNewEntry(int hashCode, K key, V value, int position) {
+ Entry<K,V> e = table[position];
+ table[position] = new Entry<>(key, value, hashCode, e);
+ numElements++;
+
+ // rehash if necessary
+ if (numElements > rehashThreshold) {
+ growTable();
+ }
+ }
+
+ private int indexOf(int hashCode) {
+ return (hashCode >> shift) & (table.length - 1);
+ }
+
+ /**
+ * Creates an iterator over the entries of this map.
+ *
+ * @return An iterator over the entries of this map.
+ */
+ @Override
+ public Iterator<Entry<K, V>> iterator() {
+ return new Iterator<Entry<K, V>>() {
+
+ private final Entry<K, V>[] tab = KeyMap.this.table;
+
+ private Entry<K, V> nextEntry;
+
+ private int nextPos = 0;
+
+ @Override
+ public boolean hasNext() {
+ if (nextEntry != null) {
+ return true;
+ }
+ else {
+ while (nextPos < tab.length) {
+ Entry<K, V> e = tab[nextPos++];
+ if (e != null) {
+ nextEntry = e;
+ return true;
+ }
+ }
+ return false;
+ }
+ }
+
+ @Override
+ public Entry<K, V> next() {
+ if (nextEntry != null || hasNext()) {
+ Entry<K, V> e = nextEntry;
+ nextEntry = nextEntry.next;
+ return e;
+ }
+ else {
+ throw new NoSuchElementException();
+ }
+ }
+
+ @Override
+ public void remove() {
+ throw new UnsupportedOperationException();
+ }
+ };
+ }
+
+ // ------------------------------------------------------------------------
+ // Properties
+ // ------------------------------------------------------------------------
+
+ /**
+ * Gets the number of elements currently in the map.
+ * @return The number of elements currently in the map.
+ */
+ public int size() {
+ return numElements;
+ }
+
+ /**
+ * Checks whether the map is empty.
+ * @return True, if the map is empty, false otherwise.
+ */
+ public boolean isEmpty() {
+ return numElements == 0;
+ }
+
+ /**
+ * Gets the current table capacity, i.e., the number of slots in the hash table, without
+ * and overflow chaining.
+ * @return The number of slots in the hash table.
+ */
+ public int getCurrentTableCapacity() {
+ return table.length;
+ }
+
+ /**
+ * Gets the base-2 logarithm of the hash table capacity, as returned by
+ * {@link #getCurrentTableCapacity()}.
+ *
+ * @return The base-2 logarithm of the hash table capacity.
+ */
+ public int getLog2TableCapacity() {
+ return log2size;
+ }
+
+ public int getRehashThreshold() {
+ return rehashThreshold;
+ }
+
+ public int getShift() {
+ return shift;
+ }
+
+ // ------------------------------------------------------------------------
+ // Utilities
+ // ------------------------------------------------------------------------
+
+ @SuppressWarnings("unchecked")
+ private Entry<K, V>[] allocateTable(int numElements) {
+ return (Entry<K, V>[]) new Entry<?, ?>[numElements];
+ }
+
+ private void growTable() {
+ final int newSize = table.length << 1;
+
+ // only grow if there is still space to grow the table
+ if (newSize > 0) {
+ final Entry<K, V>[] oldTable = this.table;
+ final Entry<K, V>[] newTable = allocateTable(newSize);
+
+ final int newShift = shift - 1;
+ final int newMask = newSize - 1;
+
+ // go over all slots from the table. since we hash to adjacent positions in
+ // the new hash table, this is actually cache efficient
+ for (Entry<K, V> entry : oldTable) {
+ // traverse the chain for each slot
+ while (entry != null) {
+ final int newPos = (entry.hashCode >> newShift) & newMask;
+ Entry<K, V> nextEntry = entry.next;
+ entry.next = newTable[newPos];
+ newTable[newPos] = entry;
+ entry = nextEntry;
+ }
+ }
+
+ this.table = newTable;
+ this.shift = newShift;
+ this.rehashThreshold = getRehashThreshold(newSize);
+ this.log2size += 1;
+ }
+ }
+
+ private static int hash(Object key) {
+ int code = key.hashCode();
+
+ // we need a strong hash function that generates diverse upper bits
+ // this hash function is more expensive than the "scramble" used by "java.util.HashMap",
+ // but required for this sort of hash table
+ code = (code + 0x7ed55d16) + (code << 12);
+ code = (code ^ 0xc761c23c) ^ (code >>> 19);
+ code = (code + 0x165667b1) + (code << 5);
+ code = (code + 0xd3a2646c) ^ (code << 9);
+ code = (code + 0xfd7046c5) + (code << 3);
+ return (code ^ 0xb55a4f09) ^ (code >>> 16);
+ }
+
+ private static int getRehashThreshold(int capacity) {
+ // divide before multiply, to avoid overflow
+ return capacity / 4 * 3;
+ }
+
+ // ------------------------------------------------------------------------
+ // Testing Utilities
+ // ------------------------------------------------------------------------
+
+ /**
+ * For testing only: Actively counts the number of entries, rather than using the
+ * counter variable. This method has linear complexity, rather than constant.
+ *
+ * @return The counted number of entries.
+ */
+ int traverseAndCountElements() {
+ int num = 0;
+
+ for (Entry<?, ?> entry : table) {
+ while (entry != null) {
+ num++;
+ entry = entry.next;
+ }
+ }
+
+ return num;
+ }
+
+ /**
+ * For testing only: Gets the length of the longest overflow chain.
+ * This method has linear complexity.
+ *
+ * @return The length of the longest overflow chain.
+ */
+ int getLongestChainLength() {
+ int maxLen = 0;
+
+ for (Entry<?, ?> entry : table) {
+ int thisLen = 0;
+ while (entry != null) {
+ thisLen++;
+ entry = entry.next;
+ }
+ maxLen = Math.max(maxLen, thisLen);
+ }
+
+ return maxLen;
+ }
+
+ // ------------------------------------------------------------------------
+
+ /**
+ * An entry in the hash table.
+ *
+ * @param <K> Type of the key.
+ * @param <V> Type of the value.
+ */
+ public static final class Entry<K, V> {
+
+ final K key;
+ final int hashCode;
+
+ V value;
+ Entry<K, V> next;
+ long touchedTag;
+
+ Entry(K key, V value, int hashCode, Entry<K, V> next) {
+ this.key = key;
+ this.value = value;
+ this.next = next;
+ this.hashCode = hashCode;
+ }
+
+ public K getKey() {
+ return key;
+ }
+
+ public V getValue() {
+ return value;
+ }
+ }
+
+ // ------------------------------------------------------------------------
+
+ /**
+ * Performs a traversal about logical the multi-map that results from the union of the
+ * given maps. This method does not actually build a union of the map, but traverses the hash maps
+ * together.
+ *
+ * @param maps The array uf maps whose union should be traversed.
+ * @param visitor The visitor that is called for each key and all values.
+ * @param touchedTag A tag that is used to mark elements that have been touched in this specific
+ * traversal. Each successive traversal should supply a larger value for this
+ * tag than the previous one.
+ *
+ * @param <K> The type of the map's key.
+ * @param <V> The type of the map's value.
+ */
+ public static <K, V> void traverseMaps(
+ final KeyMap<K, V>[] maps,
+ final TraversalEvaluator<K, V> visitor,
+ final long touchedTag)
+ throws Exception
+ {
+ // we need to work on the maps in descending size
+ Arrays.sort(maps, CapacityDescendingComparator.INSTANCE);
+
+ final int[] shifts = new int[maps.length];
+ final int[] lowBitsMask = new int[maps.length];
+ final int numSlots = maps[0].table.length;
+ final int numTables = maps.length;
+
+ // figure out how much each hash table collapses the entries
+ for (int i = 0; i < numTables; i++) {
+ shifts[i] = maps[0].log2size - maps[i].log2size;
+ lowBitsMask[i] = (1 << shifts[i]) - 1;
+ }
+
+ // go over all slots (based on the largest hash table)
+ for (int pos = 0; pos < numSlots; pos++) {
+
+ // for each slot, go over all tables, until the table does not have that slot any more
+ // for tables where multiple slots collapse into one, we visit that one when we process the
+ // latest of all slots that collapse to that one
+ int mask;
+ for (int rootTable = 0;
+ rootTable < numTables && ((mask = lowBitsMask[rootTable]) & pos) == mask;
+ rootTable++)
+ {
+ // use that table to gather keys and start collecting keys from the following tables
+ // go over all entries of that slot in the table
+ Entry<K, V> entry = maps[rootTable].table[pos >> shifts[rootTable]];
+ while (entry != null) {
+ // take only entries that have not been collected as part of other tables
+ if (entry.touchedTag < touchedTag) {
+ entry.touchedTag = touchedTag;
+
+ final K key = entry.key;
+ final int hashCode = entry.hashCode;
+ visitor.startNewKey(key);
+ visitor.nextValue(entry.value);
+
+ addEntriesFromChain(entry.next, visitor, key, touchedTag, hashCode);
+
+ // go over the other hash tables and collect their entries for the key
+ for (int followupTable = rootTable + 1; followupTable < numTables; followupTable++) {
+ Entry<K, V> followupEntry = maps[followupTable].table[pos >> shifts[followupTable]];
+ if (followupEntry != null) {
+ addEntriesFromChain(followupEntry, visitor, key, touchedTag, hashCode);
+ }
+ }
+
+ visitor.keyDone();
+ }
+
+ entry = entry.next;
+ }
+ }
+ }
+ }
+
+ private static <K, V> void addEntriesFromChain(
+ Entry<K, V> entry,
+ TraversalEvaluator<K, V> visitor,
+ K key,
+ long touchedTag,
+ int hashCode) throws Exception
+ {
+ while (entry != null) {
+ if (entry.touchedTag < touchedTag && entry.hashCode == hashCode && entry.key.equals(key)) {
+ entry.touchedTag = touchedTag;
+ visitor.nextValue(entry.value);
+ }
+ entry = entry.next;
+ }
+ }
+
+ // ------------------------------------------------------------------------
+
+ /**
+ * Comparator that defines a descending order on maps depending on their table capacity
+ * and number of elements.
+ */
+ static final class CapacityDescendingComparator implements Comparator<KeyMap<?, ?>> {
+
+ static final CapacityDescendingComparator INSTANCE = new CapacityDescendingComparator();
+
+ private CapacityDescendingComparator() {}
+
+
+ @Override
+ public int compare(KeyMap<?, ?> o1, KeyMap<?, ?> o2) {
+ // this sorts descending
+ int cmp = o2.getLog2TableCapacity() - o1.getLog2TableCapacity();
+ if (cmp != 0) {
+ return cmp;
+ }
+ else {
+ return o2.size() - o1.size();
+ }
+ }
+ }
+
+ // ------------------------------------------------------------------------
+
+ /**
+ * A factory for lazy/on-demand instantiation of values.
+ *
+ * @param <V> The type created by the factory.
+ */
+ public static interface LazyFactory<V> {
+
+ /**
+ * The factory method; creates the value.
+ * @return The value.
+ */
+ V create();
+ }
+
+ // ------------------------------------------------------------------------
+
+ /**
+ * A visitor for a traversal over the union of multiple hash maps. The visitor is
+ * called for each key in the union of the maps and all values associated with that key
+ * (one per map, but multiple across maps).
+ *
+ * @param <K> The type of the key.
+ * @param <V> The type of the value.
+ */
+ public static interface TraversalEvaluator<K, V> {
+
+ /**
+ * Called whenever the traversal starts with a new key.
+ *
+ * @param key The key traversed.
+ * @throws Exception Method forwards all exceptions.
+ */
+ void startNewKey(K key) throws Exception;
+
+ /**
+ * Called for each value found for the current key.
+ *
+ * @param value The next value.
+ * @throws Exception Method forwards all exceptions.
+ */
+ void nextValue(V value) throws Exception;
+
+ /**
+ * Called when the traversal for the current key is complete.
+ *
+ * @throws Exception Method forwards all exceptions.
+ */
+ void keyDone() throws Exception;
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/b8f58fab/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/operators/windows/package-info.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/operators/windows/package-info.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/operators/windows/package-info.java
new file mode 100644
index 0000000..63ed470
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/operators/windows/package-info.java
@@ -0,0 +1,22 @@
+/*
+ * 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.
+ */
+
+/**
+ * This package contains the operators that implement the various window operations
+ * on data streams.
+ */
+package org.apache.flink.streaming.runtime.operators.windows;
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/b8f58fab/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTaskException.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTaskException.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTaskException.java
index d93078b..5680810 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTaskException.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTaskException.java
@@ -18,7 +18,7 @@
package org.apache.flink.streaming.runtime.tasks;
/**
- * An exception that is thrown by the stream verices when encountering an
+ * An exception that is thrown by the stream vertices when encountering an
* illegal condition.
*/
public class StreamTaskException extends RuntimeException {
http://git-wip-us.apache.org/repos/asf/flink/blob/b8f58fab/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/package-info.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/package-info.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/package-info.java
new file mode 100644
index 0000000..a40ae3a
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/package-info.java
@@ -0,0 +1,27 @@
+/*
+ * 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.
+ */
+
+/**
+ * This package contains classes that realize streaming tasks. These tasks are
+ * executable stream consumers and producers that are scheduled by the distributed
+ * dataflow runtime. Each task occupies one execution slot and is run with by an
+ * executing thread.
+ * <p>
+ * The tasks merely set up the distributed stream coordination and the checkpointing.
+ * Internally, the tasks create one or more operators, perform the stream transformations.
+ */
+package org.apache.flink.streaming.runtime.tasks;
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/b8f58fab/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/runtime/operators/windows/AccumulatingAlignedProcessingTimeWindowOperatorTest.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/runtime/operators/windows/AccumulatingAlignedProcessingTimeWindowOperatorTest.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/runtime/operators/windows/AccumulatingAlignedProcessingTimeWindowOperatorTest.java
new file mode 100644
index 0000000..685939b
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/runtime/operators/windows/AccumulatingAlignedProcessingTimeWindowOperatorTest.java
@@ -0,0 +1,515 @@
+/*
+ * 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.runtime.operators.windows;
+
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.functions.windows.KeyedWindowFunction;
+import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.runtime.operators.TriggerTimer;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.StreamingRuntimeContext;
+
+import org.apache.flink.util.Collector;
+import org.junit.After;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import static org.mockito.Mockito.*;
+import static org.junit.Assert.*;
+
+@SuppressWarnings("serial")
+public class AccumulatingAlignedProcessingTimeWindowOperatorTest {
+
+ @SuppressWarnings("unchecked")
+ private final KeyedWindowFunction<String, String, String> mockFunction = mock(KeyedWindowFunction.class);
+
+ @SuppressWarnings("unchecked")
+ private final KeySelector<String, String> mockKeySelector = mock(KeySelector.class);
+
+ private final KeySelector<Integer, Integer> identitySelector = new KeySelector<Integer, Integer>() {
+ @Override
+ public Integer getKey(Integer value) {
+ return value;
+ }
+ };
+
+ private final KeyedWindowFunction<Integer, Integer, Integer> validatingIdentityFunction =
+ new KeyedWindowFunction<Integer, Integer, Integer>()
+ {
+ @Override
+ public void evaluate(Integer key, Iterable<Integer> values, Collector<Integer> out) {
+ for (Integer val : values) {
+ assertEquals(key, val);
+ out.collect(val);
+ }
+ }
+ };
+
+ // ------------------------------------------------------------------------
+
+ @After
+ public void checkNoTriggerThreadsRunning() {
+ // make sure that all the threads we trigger are shut down
+ long deadline = System.currentTimeMillis() + 5000;
+ while (TriggerTimer.TRIGGER_THREADS_GROUP.activeCount() > 0 && System.currentTimeMillis() < deadline) {
+ try {
+ Thread.sleep(10);
+ }
+ catch (InterruptedException ignored) {}
+ }
+
+ assertTrue("Not all trigger threads where properly shut down",
+ TriggerTimer.TRIGGER_THREADS_GROUP.activeCount() == 0);
+ }
+
+ // ------------------------------------------------------------------------
+
+ @Test
+ public void testInvalidParameters() {
+ try {
+ assertInvalidParameter(-1L, -1L);
+ assertInvalidParameter(10000L, -1L);
+ assertInvalidParameter(-1L, 1000L);
+ assertInvalidParameter(1000L, 2000L);
+
+ // actual internal slide is too low here:
+ assertInvalidParameter(1000L, 999L);
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testWindowSizeAndSlide() {
+ try {
+ AbstractAlignedProcessingTimeWindowOperator<String, String, String> op;
+
+ op = new AccumulatingProcessingTimeWindowOperator<>(mockFunction, mockKeySelector, 5000, 1000);
+ assertEquals(5000, op.getWindowSize());
+ assertEquals(1000, op.getWindowSlide());
+ assertEquals(1000, op.getPaneSize());
+ assertEquals(5, op.getNumPanesPerWindow());
+
+ op = new AccumulatingProcessingTimeWindowOperator<>(mockFunction, mockKeySelector, 1000, 1000);
+ assertEquals(1000, op.getWindowSize());
+ assertEquals(1000, op.getWindowSlide());
+ assertEquals(1000, op.getPaneSize());
+ assertEquals(1, op.getNumPanesPerWindow());
+
+ op = new AccumulatingProcessingTimeWindowOperator<>(mockFunction, mockKeySelector, 1500, 1000);
+ assertEquals(1500, op.getWindowSize());
+ assertEquals(1000, op.getWindowSlide());
+ assertEquals(500, op.getPaneSize());
+ assertEquals(3, op.getNumPanesPerWindow());
+
+ op = new AccumulatingProcessingTimeWindowOperator<>(mockFunction, mockKeySelector, 1200, 1100);
+ assertEquals(1200, op.getWindowSize());
+ assertEquals(1100, op.getWindowSlide());
+ assertEquals(100, op.getPaneSize());
+ assertEquals(12, op.getNumPanesPerWindow());
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testWindowTriggerTimeAlignment() {
+ try {
+ @SuppressWarnings("unchecked")
+ final Output<StreamRecord<String>> mockOut = mock(Output.class);
+
+ final StreamingRuntimeContext mockContext = mock(StreamingRuntimeContext.class);
+ when(mockContext.getTaskName()).thenReturn("Test task name");
+
+ AbstractAlignedProcessingTimeWindowOperator<String, String, String> op;
+
+ op = new AccumulatingProcessingTimeWindowOperator<>(mockFunction, mockKeySelector, 5000, 1000);
+ op.setup(mockOut, mockContext);
+ op.open(new Configuration());
+ assertTrue(op.getNextSlideTime() % 1000 == 0);
+ assertTrue(op.getNextEvaluationTime() % 1000 == 0);
+ op.dispose();
+
+ op = new AccumulatingProcessingTimeWindowOperator<>(mockFunction, mockKeySelector, 1000, 1000);
+ op.setup(mockOut, mockContext);
+ op.open(new Configuration());
+ assertTrue(op.getNextSlideTime() % 1000 == 0);
+ assertTrue(op.getNextEvaluationTime() % 1000 == 0);
+ op.dispose();
+
+ op = new AccumulatingProcessingTimeWindowOperator<>(mockFunction, mockKeySelector, 1500, 1000);
+ op.setup(mockOut, mockContext);
+ op.open(new Configuration());
+ assertTrue(op.getNextSlideTime() % 500 == 0);
+ assertTrue(op.getNextEvaluationTime() % 1000 == 0);
+ op.dispose();
+
+ op = new AccumulatingProcessingTimeWindowOperator<>(mockFunction, mockKeySelector, 1200, 1100);
+ op.setup(mockOut, mockContext);
+ op.open(new Configuration());
+ assertTrue(op.getNextSlideTime() % 100 == 0);
+ assertTrue(op.getNextEvaluationTime() % 1100 == 0);
+ op.dispose();
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testTumblingWindow() {
+ try {
+ final int windowSize = 50;
+ final CollectingOutput<Integer> out = new CollectingOutput<>(windowSize);
+
+ final StreamingRuntimeContext mockContext = mock(StreamingRuntimeContext.class);
+ when(mockContext.getTaskName()).thenReturn("Test task name");
+
+ // tumbling window that triggers every 20 milliseconds
+ AbstractAlignedProcessingTimeWindowOperator<Integer, Integer, Integer> op =
+ new AccumulatingProcessingTimeWindowOperator<>(
+ validatingIdentityFunction, identitySelector, windowSize, windowSize);
+
+ op.setup(out, mockContext);
+ op.open(new Configuration());
+
+ final int numElements = 1000;
+
+ for (int i = 0; i < numElements; i++) {
+ op.processElement(new StreamRecord<Integer>(i));
+ Thread.sleep(1);
+ }
+
+ op.close();
+ op.dispose();
+
+ // get and verify the result
+ List<Integer> result = out.getElements();
+ assertEquals(numElements, result.size());
+
+ Collections.sort(result);
+ for (int i = 0; i < numElements; i++) {
+ assertEquals(i, result.get(i).intValue());
+ }
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testSlidingWindow() {
+ try {
+ final CollectingOutput<Integer> out = new CollectingOutput<>(50);
+
+ final StreamingRuntimeContext mockContext = mock(StreamingRuntimeContext.class);
+ when(mockContext.getTaskName()).thenReturn("Test task name");
+
+ // tumbling window that triggers every 20 milliseconds
+ AbstractAlignedProcessingTimeWindowOperator<Integer, Integer, Integer> op =
+ new AccumulatingProcessingTimeWindowOperator<>(validatingIdentityFunction, identitySelector, 150, 50);
+
+ op.setup(out, mockContext);
+ op.open(new Configuration());
+
+ final int numElements = 1000;
+
+ for (int i = 0; i < numElements; i++) {
+ op.processElement(new StreamRecord<Integer>(i));
+ Thread.sleep(1);
+ }
+
+ op.close();
+ op.dispose();
+
+ // get and verify the result
+ List<Integer> result = out.getElements();
+
+ // if we kept this running, each element would be in the result three times (for each slide).
+ // we are closing the window before the final panes are through three times, so we may have less
+ // elements.
+ if (result.size() < numElements || result.size() > 3 * numElements) {
+ fail("Wrong number of results: " + result.size());
+ }
+
+ Collections.sort(result);
+ int lastNum = -1;
+ int lastCount = -1;
+
+ for (int num : result) {
+ if (num == lastNum) {
+ lastCount++;
+ assertTrue(lastCount <= 3);
+ }
+ else {
+ lastNum = num;
+ lastCount = 1;
+ }
+ }
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testTumblingWindowSingleElements() {
+ try {
+ final CollectingOutput<Integer> out = new CollectingOutput<>(50);
+
+ final StreamingRuntimeContext mockContext = mock(StreamingRuntimeContext.class);
+ when(mockContext.getTaskName()).thenReturn("Test task name");
+
+ // tumbling window that triggers every 20 milliseconds
+ AbstractAlignedProcessingTimeWindowOperator<Integer, Integer, Integer> op =
+ new AccumulatingProcessingTimeWindowOperator<>(validatingIdentityFunction, identitySelector, 50, 50);
+
+ op.setup(out, mockContext);
+ op.open(new Configuration());
+
+ op.processElement(new StreamRecord<Integer>(1));
+ op.processElement(new StreamRecord<Integer>(2));
+ out.waitForNElements(2, 60000);
+
+ op.processElement(new StreamRecord<Integer>(3));
+ op.processElement(new StreamRecord<Integer>(4));
+ op.processElement(new StreamRecord<Integer>(5));
+ out.waitForNElements(5, 60000);
+
+ op.processElement(new StreamRecord<Integer>(6));
+ out.waitForNElements(6, 60000);
+
+ List<Integer> result = out.getElements();
+ assertEquals(6, result.size());
+
+ Collections.sort(result);
+ assertEquals(Arrays.asList(1, 2, 3, 4, 5, 6), result);
+
+ op.close();
+ op.dispose();
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testSlidingWindowSingleElements() {
+ try {
+ final CollectingOutput<Integer> out = new CollectingOutput<>(50);
+
+ final StreamingRuntimeContext mockContext = mock(StreamingRuntimeContext.class);
+ when(mockContext.getTaskName()).thenReturn("Test task name");
+
+ // tumbling window that triggers every 20 milliseconds
+ AbstractAlignedProcessingTimeWindowOperator<Integer, Integer, Integer> op =
+ new AccumulatingProcessingTimeWindowOperator<>(validatingIdentityFunction, identitySelector, 150, 50);
+
+ op.setup(out, mockContext);
+ op.open(new Configuration());
+
+ op.processElement(new StreamRecord<Integer>(1));
+ op.processElement(new StreamRecord<Integer>(2));
+
+ // each element should end up in the output three times
+ // wait until the elements have arrived 6 times in the output
+ out.waitForNElements(6, 120000);
+
+ List<Integer> result = out.getElements();
+ assertEquals(6, result.size());
+
+ Collections.sort(result);
+ assertEquals(Arrays.asList(1, 1, 1, 2, 2, 2), result);
+
+ op.close();
+ op.dispose();
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testEmitTrailingDataOnClose() {
+ try {
+ final CollectingOutput<Integer> out = new CollectingOutput<>();
+
+ final StreamingRuntimeContext mockContext = mock(StreamingRuntimeContext.class);
+ when(mockContext.getTaskName()).thenReturn("Test task name");
+
+ // the operator has a window time that is so long that it will not fire in this test
+ final long oneYear = 365L * 24 * 60 * 60 * 1000;
+ AbstractAlignedProcessingTimeWindowOperator<Integer, Integer, Integer> op =
+ new AccumulatingProcessingTimeWindowOperator<>(validatingIdentityFunction, identitySelector,
+ oneYear, oneYear);
+
+ op.setup(out, mockContext);
+ op.open(new Configuration());
+
+ List<Integer> data = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
+ for (Integer i : data) {
+ op.processElement(new StreamRecord<Integer>(i));
+ }
+
+ op.close();
+ op.dispose();
+
+ // get and verify the result
+ List<Integer> result = out.getElements();
+ Collections.sort(result);
+ assertEquals(data, result);
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testPropagateExceptionsFromTrigger() {
+ try {
+ final CollectingOutput<Integer> out = new CollectingOutput<>();
+
+ final StreamingRuntimeContext mockContext = mock(StreamingRuntimeContext.class);
+ when(mockContext.getTaskName()).thenReturn("Test task name");
+
+ KeyedWindowFunction<Integer, Integer, Integer> failingFunction = new FailingFunction(100);
+
+ AbstractAlignedProcessingTimeWindowOperator<Integer, Integer, Integer> op =
+ new AccumulatingProcessingTimeWindowOperator<>(failingFunction, identitySelector, 50, 50);
+
+ op.setup(out, mockContext);
+ op.open(new Configuration());
+
+ try {
+ int num = 0;
+ while (num < Integer.MAX_VALUE) {
+ op.processElement(new StreamRecord<Integer>(num++));
+ Thread.sleep(1);
+ }
+ fail("This should really have failed with an exception quite a while ago...");
+ }
+ catch (Exception e) {
+ assertNotNull(e.getCause());
+ assertTrue(e.getCause().getMessage().contains("Artificial Test Exception"));
+ }
+
+ op.dispose();
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testPropagateExceptionsFromClose() {
+ try {
+ final CollectingOutput<Integer> out = new CollectingOutput<>();
+
+ final StreamingRuntimeContext mockContext = mock(StreamingRuntimeContext.class);
+ when(mockContext.getTaskName()).thenReturn("Test task name");
+
+ KeyedWindowFunction<Integer, Integer, Integer> failingFunction = new FailingFunction(100);
+
+ // the operator has a window time that is so long that it will not fire in this test
+ final long hundredYears = 100L * 365 * 24 * 60 * 60 * 1000;
+ AbstractAlignedProcessingTimeWindowOperator<Integer, Integer, Integer> op =
+ new AccumulatingProcessingTimeWindowOperator<>(
+ failingFunction, identitySelector, hundredYears, hundredYears);
+
+ op.setup(out, mockContext);
+ op.open(new Configuration());
+
+ for (int i = 0; i < 150; i++) {
+ op.processElement(new StreamRecord<Integer>(i));
+ }
+
+ try {
+ op.close();
+ fail("This should fail with an exception");
+ }
+ catch (Exception e) {
+ assertTrue(
+ e.getMessage().contains("Artificial Test Exception") ||
+ (e.getCause() != null && e.getCause().getMessage().contains("Artificial Test Exception")));
+ }
+
+ op.dispose();
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ // ------------------------------------------------------------------------
+
+ private void assertInvalidParameter(long windowSize, long windowSlide) {
+ try {
+ new AccumulatingProcessingTimeWindowOperator<String, String, String>(
+ mockFunction, mockKeySelector, windowSize, windowSlide);
+ fail("This should fail with an IllegalArgumentException");
+ }
+ catch (IllegalArgumentException e) {
+ // expected
+ }
+ catch (Exception e) {
+ fail("Wrong exception. Expected IllegalArgumentException but found " + e.getClass().getSimpleName());
+ }
+ }
+
+ // ------------------------------------------------------------------------
+
+ private static class FailingFunction implements KeyedWindowFunction<Integer, Integer, Integer> {
+
+ private final int failAfterElements;
+
+ private int numElements;
+
+ FailingFunction(int failAfterElements) {
+ this.failAfterElements = failAfterElements;
+ }
+
+ @Override
+ public void evaluate(Integer integer, Iterable<Integer> values, Collector<Integer> out) throws Exception {
+ for (Integer i : values) {
+ out.collect(i);
+ numElements++;
+
+ if (numElements >= failAfterElements) {
+ throw new Exception("Artificial Test Exception");
+ }
+ }
+ }
+ }
+}