flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From rmetz...@apache.org
Subject [12/52] [partial] flink git commit: [FLINK-1452] Rename 'flink-addons' to 'flink-staging'
Date Mon, 02 Feb 2015 18:41:50 GMT
http://git-wip-us.apache.org/repos/asf/flink/blob/f94112fb/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/function/source/FromElementsFunction.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/function/source/FromElementsFunction.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/function/source/FromElementsFunction.java
new file mode 100755
index 0000000..8afac75
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/function/source/FromElementsFunction.java
@@ -0,0 +1,49 @@
+/*
+ * 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.function.source;
+
+import java.util.Arrays;
+import java.util.Collection;
+
+import org.apache.flink.util.Collector;
+
+public class FromElementsFunction<T> implements SourceFunction<T> {
+	private static final long serialVersionUID = 1L;
+
+	Iterable<T> iterable;
+
+	public FromElementsFunction(T... elements) {
+		this.iterable = Arrays.asList(elements);
+	}
+
+	public FromElementsFunction(Collection<T> elements) {
+		this.iterable = elements;
+	}
+
+	public FromElementsFunction(Iterable<T> elements) {
+		this.iterable = elements;
+	}
+
+	@Override
+	public void invoke(Collector<T> collector) throws Exception {
+		for (T element : iterable) {
+			collector.collect(element);
+		}
+	}
+
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/f94112fb/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/function/source/GenSequenceFunction.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/function/source/GenSequenceFunction.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/function/source/GenSequenceFunction.java
new file mode 100755
index 0000000..3afd06e
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/function/source/GenSequenceFunction.java
@@ -0,0 +1,53 @@
+/*
+ * 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.function.source;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.util.Collector;
+import org.apache.flink.util.NumberSequenceIterator;
+
+/**
+ * Source Function used to generate the number sequence
+ * 
+ */
+public class GenSequenceFunction extends RichParallelSourceFunction<Long> {
+
+	private static final long serialVersionUID = 1L;
+
+	private NumberSequenceIterator fullIterator;
+	private NumberSequenceIterator splitIterator;
+
+	public GenSequenceFunction(long from, long to) {
+		fullIterator = new NumberSequenceIterator(from, to);
+	}
+
+	@Override
+	public void invoke(Collector<Long> collector) throws Exception {
+		while (splitIterator.hasNext()) {
+			collector.collect(splitIterator.next());
+		}
+	}
+
+	@Override
+	public void open(Configuration config) {
+		int splitNumber = getRuntimeContext().getIndexOfThisSubtask();
+		int numOfSubTasks = getRuntimeContext().getNumberOfParallelSubtasks();
+		splitIterator = fullIterator.split(numOfSubTasks)[splitNumber];
+	}
+
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/f94112fb/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/function/source/GenericSourceFunction.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/function/source/GenericSourceFunction.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/function/source/GenericSourceFunction.java
new file mode 100644
index 0000000..664d39a
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/function/source/GenericSourceFunction.java
@@ -0,0 +1,25 @@
+/*
+ * 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.function.source;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+
+public interface GenericSourceFunction<T> {
+
+	public TypeInformation<T> getType();
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/f94112fb/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/function/source/ParallelSourceFunction.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/function/source/ParallelSourceFunction.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/function/source/ParallelSourceFunction.java
new file mode 100644
index 0000000..46d4fe9
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/function/source/ParallelSourceFunction.java
@@ -0,0 +1,26 @@
+/*
+ * 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.function.source;
+
+import org.apache.flink.util.Collector;
+
+public interface ParallelSourceFunction<OUT> extends SourceFunction<OUT> {
+
+	public void invoke(Collector<OUT> collector) throws Exception;
+		
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/f94112fb/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/function/source/RichParallelSourceFunction.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/function/source/RichParallelSourceFunction.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/function/source/RichParallelSourceFunction.java
new file mode 100644
index 0000000..5bbfd4c
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/function/source/RichParallelSourceFunction.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.api.function.source;
+
+import org.apache.flink.api.common.functions.AbstractRichFunction;
+
+public abstract class RichParallelSourceFunction<OUT> extends AbstractRichFunction implements
+		ParallelSourceFunction<OUT> {
+
+	private static final long serialVersionUID = 1L;
+
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/f94112fb/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/function/source/RichSourceFunction.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/function/source/RichSourceFunction.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/function/source/RichSourceFunction.java
new file mode 100755
index 0000000..4b947c7
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/function/source/RichSourceFunction.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.api.function.source;
+
+import org.apache.flink.api.common.functions.AbstractRichFunction;
+
+public abstract class RichSourceFunction<OUT> extends AbstractRichFunction implements
+		SourceFunction<OUT> {
+
+	private static final long serialVersionUID = 1L;
+
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/f94112fb/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/function/source/SocketTextStreamFunction.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/function/source/SocketTextStreamFunction.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/function/source/SocketTextStreamFunction.java
new file mode 100644
index 0000000..ac82b10
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/function/source/SocketTextStreamFunction.java
@@ -0,0 +1,86 @@
+/*
+ * 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.function.source;
+
+import java.io.BufferedReader;
+import java.io.InputStreamReader;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.util.Collector;
+
+public class SocketTextStreamFunction extends RichSourceFunction<String> {
+	private static final long serialVersionUID = 1L;
+	
+	private String hostname;
+	private int port;
+	private char delimiter;
+	private Socket socket;
+	private static final int CONNECTION_TIMEOUT_TIME = 0;
+
+	public SocketTextStreamFunction(String hostname, int port, char delimiter) {
+		this.hostname = hostname;
+		this.port = port;
+		this.delimiter = delimiter;
+	}
+
+	@Override
+	public void open(Configuration parameters) throws Exception {
+		super.open(parameters);
+		socket = new Socket();
+		
+		socket.connect(new InetSocketAddress(hostname, port), CONNECTION_TIMEOUT_TIME);
+	}
+	
+	@Override
+	public void invoke(Collector<String> collector) throws Exception {
+		while (!socket.isClosed() && socket.isConnected()) {
+			streamFromSocket(collector, socket);
+		}
+	}
+
+	public void streamFromSocket(Collector<String> collector, Socket socket) throws Exception {
+		StringBuffer buffer = new StringBuffer();
+		BufferedReader reader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
+
+		while (true) {
+			int data = reader.read();
+			if (!socket.isConnected() || socket.isClosed() || data == -1) {
+				break;
+			}
+
+			if (data == delimiter) {
+				collector.collect(buffer.toString());
+				buffer = new StringBuffer();
+			} else if (data != '\r') { // ignore carriage return
+				buffer.append((char) data);
+			}
+		}
+
+		if (buffer.length() > 0) {
+			collector.collect(buffer.toString());
+		}
+	}
+
+	@Override
+	public void close() throws Exception {
+		socket.close();
+		super.close();
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/f94112fb/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/function/source/SourceFunction.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/function/source/SourceFunction.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/function/source/SourceFunction.java
new file mode 100755
index 0000000..917562a
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/function/source/SourceFunction.java
@@ -0,0 +1,29 @@
+/*
+ * 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.function.source;
+
+import java.io.Serializable;
+
+import org.apache.flink.api.common.functions.Function;
+import org.apache.flink.util.Collector;
+
+public interface SourceFunction<OUT> extends Function, Serializable {
+
+	public void invoke(Collector<OUT> collector) throws Exception;
+		
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/f94112fb/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/ChainableInvokable.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/ChainableInvokable.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/ChainableInvokable.java
new file mode 100644
index 0000000..24c0319
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/ChainableInvokable.java
@@ -0,0 +1,39 @@
+/*
+ * 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.invokable;
+
+import org.apache.flink.api.common.functions.Function;
+import org.apache.flink.streaming.api.streamrecord.StreamRecordSerializer;
+import org.apache.flink.util.Collector;
+
+public abstract class ChainableInvokable<IN, OUT> extends StreamInvokable<IN, OUT> implements
+		Collector<IN> {
+
+	private static final long serialVersionUID = 1L;
+
+	public ChainableInvokable(Function userFunction) {
+		super(userFunction);
+		setChainingStrategy(ChainingStrategy.ALWAYS);
+	}
+
+	public void setup(Collector<OUT> collector, StreamRecordSerializer<IN> inSerializer) {
+		this.collector = collector;
+		this.inSerializer = inSerializer;
+		this.objectSerializer = inSerializer.getObjectSerializer();
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/f94112fb/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/SinkInvokable.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/SinkInvokable.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/SinkInvokable.java
new file mode 100755
index 0000000..13a6ba1
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/SinkInvokable.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.streaming.api.invokable;
+
+import org.apache.flink.streaming.api.function.sink.SinkFunction;
+
+public class SinkInvokable<IN> extends ChainableInvokable<IN, IN> {
+	private static final long serialVersionUID = 1L;
+
+	private SinkFunction<IN> sinkFunction;
+
+	public SinkInvokable(SinkFunction<IN> sinkFunction) {
+		super(sinkFunction);
+		this.sinkFunction = sinkFunction;
+	}
+
+	@Override
+	public void invoke() throws Exception {
+		while (readNext() != null) {
+			callUserFunctionAndLogException();
+		}
+	}
+
+	@Override
+	protected void callUserFunction() throws Exception {
+		sinkFunction.invoke(nextObject);
+	}
+
+	@Override
+	public void collect(IN record) {
+		nextObject = copy(record);
+		callUserFunctionAndLogException();
+	}
+
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/f94112fb/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/SourceInvokable.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/SourceInvokable.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/SourceInvokable.java
new file mode 100644
index 0000000..f1cf2c5
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/SourceInvokable.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.invokable;
+
+import java.io.Serializable;
+
+import org.apache.flink.streaming.api.function.source.SourceFunction;
+
+public class SourceInvokable<OUT> extends StreamInvokable<OUT, OUT> implements Serializable {
+
+	private static final long serialVersionUID = 1L;
+
+	private SourceFunction<OUT> sourceFunction;
+
+	public SourceInvokable(SourceFunction<OUT> sourceFunction) {
+		super(sourceFunction);
+		this.sourceFunction = sourceFunction;
+	}
+
+	@Override
+	public void invoke() {
+		callUserFunctionAndLogException();
+	}
+
+	@Override
+	protected void callUserFunction() throws Exception {
+		sourceFunction.invoke(collector);
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/f94112fb/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/StreamInvokable.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/StreamInvokable.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/StreamInvokable.java
new file mode 100644
index 0000000..6cee5f2
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/StreamInvokable.java
@@ -0,0 +1,186 @@
+/*
+ * 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.invokable;
+
+import java.io.IOException;
+import java.io.Serializable;
+
+import org.apache.flink.api.common.functions.Function;
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.api.common.functions.util.FunctionUtils;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.streamrecord.StreamRecord;
+import org.apache.flink.streaming.api.streamrecord.StreamRecordSerializer;
+import org.apache.flink.streaming.api.streamvertex.StreamTaskContext;
+import org.apache.flink.util.Collector;
+import org.apache.flink.util.MutableObjectIterator;
+import org.apache.flink.util.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * The StreamInvokable represents the base class for all invokables in the
+ * streaming topology.
+ * 
+ * @param <OUT>
+ *            The output type of the invokable
+ */
+public abstract class StreamInvokable<IN, OUT> implements Serializable {
+
+	private static final long serialVersionUID = 1L;
+	private static final Logger LOG = LoggerFactory.getLogger(StreamInvokable.class);
+
+	protected StreamTaskContext<OUT> taskContext;
+
+	protected MutableObjectIterator<StreamRecord<IN>> recordIterator;
+	protected StreamRecordSerializer<IN> inSerializer;
+	protected TypeSerializer<IN> objectSerializer;
+	protected StreamRecord<IN> nextRecord;
+	protected IN nextObject;
+	protected boolean isMutable;
+
+	protected Collector<OUT> collector;
+	protected Function userFunction;
+	protected volatile boolean isRunning;
+
+	private ChainingStrategy chainingStrategy = ChainingStrategy.HEAD;
+
+	public StreamInvokable(Function userFunction) {
+		this.userFunction = userFunction;
+	}
+
+	/**
+	 * Initializes the {@link StreamInvokable} for input and output handling
+	 * 
+	 * @param taskContext
+	 *            StreamTaskContext representing the vertex
+	 */
+	public void setup(StreamTaskContext<OUT> taskContext) {
+		this.collector = taskContext.getOutputCollector();
+		this.recordIterator = taskContext.getInput(0);
+		this.inSerializer = taskContext.getInputSerializer(0);
+		if (this.inSerializer != null) {
+			this.nextRecord = inSerializer.createInstance();
+			this.objectSerializer = inSerializer.getObjectSerializer();
+		}
+		this.taskContext = taskContext;
+	}
+
+	/**
+	 * Method that will be called when the operator starts, should encode the
+	 * processing logic
+	 */
+	public abstract void invoke() throws Exception;
+
+	/*
+	 * Reads the next record from the reader iterator and stores it in the
+	 * nextRecord variable
+	 */
+	protected StreamRecord<IN> readNext() {
+		this.nextRecord = inSerializer.createInstance();
+		try {
+			nextRecord = recordIterator.next(nextRecord);
+			try {
+				nextObject = nextRecord.getObject();
+			} catch (NullPointerException e) {
+				// end of stream
+			}
+			return nextRecord;
+		} catch (IOException e) {
+			throw new RuntimeException("Could not read next record.");
+		}
+	}
+
+	/**
+	 * The call of the user implemented function should be implemented here
+	 */
+	protected void callUserFunction() throws Exception {
+	}
+
+	/**
+	 * Method for logging exceptions thrown during the user function call
+	 */
+	protected void callUserFunctionAndLogException() {
+		try {
+			callUserFunction();
+		} catch (Exception e) {
+			if (LOG.isErrorEnabled()) {
+				LOG.error("Calling user function failed due to: {}",
+						StringUtils.stringifyException(e));
+			}
+		}
+	}
+
+	/**
+	 * Open method to be used if the user defined function extends the
+	 * RichFunction class
+	 * 
+	 * @param parameters
+	 *            The configuration parameters for the operator
+	 */
+	public void open(Configuration parameters) throws Exception {
+		isRunning = true;
+		FunctionUtils.openFunction(userFunction, parameters);
+	}
+
+	/**
+	 * Close method to be used if the user defined function extends the
+	 * RichFunction class
+	 * 
+	 */
+	public void close() {
+		isRunning = false;
+		collector.close();
+		try {
+			FunctionUtils.closeFunction(userFunction);
+		} catch (Exception e) {
+			throw new RuntimeException("Error when closing the function: " + e.getMessage());
+		}
+	}
+
+	public void setRuntimeContext(RuntimeContext t) {
+		FunctionUtils.setFunctionRuntimeContext(userFunction, t);
+	}
+
+	protected IN copy(IN record) {
+		return objectSerializer.copy(record);
+	}
+
+	public void setChainingStrategy(ChainingStrategy strategy) {
+		if (strategy == ChainingStrategy.ALWAYS) {
+			if (!(this instanceof ChainableInvokable)) {
+				throw new RuntimeException(
+						"Invokable needs to extend ChainableInvokable to be chained");
+			}
+		}
+		this.chainingStrategy = strategy;
+	}
+
+	public ChainingStrategy getChainingStrategy() {
+		return chainingStrategy;
+	}
+
+	public static enum ChainingStrategy {
+		ALWAYS, NEVER, HEAD;
+	}
+
+	public Function getUserFunction() {
+		return userFunction;
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/f94112fb/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/BatchIterator.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/BatchIterator.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/BatchIterator.java
new file mode 100755
index 0000000..c3475e9
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/BatchIterator.java
@@ -0,0 +1,25 @@
+/*
+ * 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.invokable.operator;
+
+import java.io.Serializable;
+import java.util.Iterator;
+
+public interface BatchIterator<IN> extends Iterator<IN>, Serializable {
+	public void reset();
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/f94112fb/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/CounterInvokable.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/CounterInvokable.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/CounterInvokable.java
new file mode 100644
index 0000000..3fc314c
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/CounterInvokable.java
@@ -0,0 +1,43 @@
+/*
+ * 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.invokable.operator;
+
+import org.apache.flink.streaming.api.invokable.ChainableInvokable;
+
+public class CounterInvokable<IN> extends ChainableInvokable<IN, Long> {
+	private static final long serialVersionUID = 1L;
+
+	Long count = 0L;
+
+	public CounterInvokable() {
+		super(null);
+	}
+
+	@Override
+	public void invoke() throws Exception {
+		while (readNext() != null) {
+			collector.collect(++count);
+		}
+	}
+
+	@Override
+	public void collect(IN record) {
+		collector.collect(++count);
+	}
+
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/f94112fb/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/FilterInvokable.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/FilterInvokable.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/FilterInvokable.java
new file mode 100644
index 0000000..0c8298e
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/FilterInvokable.java
@@ -0,0 +1,55 @@
+/*
+ * 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.invokable.operator;
+
+import org.apache.flink.api.common.functions.FilterFunction;
+import org.apache.flink.streaming.api.invokable.ChainableInvokable;
+
+public class FilterInvokable<IN> extends ChainableInvokable<IN, IN> {
+
+	private static final long serialVersionUID = 1L;
+
+	FilterFunction<IN> filterFunction;
+	private boolean collect;
+
+	public FilterInvokable(FilterFunction<IN> filterFunction) {
+		super(filterFunction);
+		this.filterFunction = filterFunction;
+	}
+
+	@Override
+	public void invoke() throws Exception {
+		while (readNext() != null) {
+			callUserFunctionAndLogException();
+		}
+	}
+
+	@Override
+	protected void callUserFunction() throws Exception {
+		collect = filterFunction.filter(copy(nextObject));
+		if (collect) {
+			collector.collect(nextObject);
+		}
+	}
+
+	@Override
+	public void collect(IN record) {
+		nextObject = copy(record);
+		callUserFunctionAndLogException();
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/f94112fb/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/FlatMapInvokable.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/FlatMapInvokable.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/FlatMapInvokable.java
new file mode 100644
index 0000000..2a4081b
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/FlatMapInvokable.java
@@ -0,0 +1,51 @@
+/*
+ * 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.invokable.operator;
+
+import org.apache.flink.api.common.functions.FlatMapFunction;
+import org.apache.flink.streaming.api.invokable.ChainableInvokable;
+
+public class FlatMapInvokable<IN, OUT> extends ChainableInvokable<IN, OUT> {
+	private static final long serialVersionUID = 1L;
+
+	private FlatMapFunction<IN, OUT> flatMapper;
+
+	public FlatMapInvokable(FlatMapFunction<IN, OUT> flatMapper) {
+		super(flatMapper);
+		this.flatMapper = flatMapper;
+	}
+
+	@Override
+	public void invoke() throws Exception {
+		while (readNext() != null) {
+			callUserFunctionAndLogException();
+		}
+	}
+
+	@Override
+	protected void callUserFunction() throws Exception {
+		flatMapper.flatMap(nextObject, collector);
+	}
+
+	@Override
+	public void collect(IN record) {
+		nextObject = copy(record);
+		callUserFunctionAndLogException();
+	}
+
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/f94112fb/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/GroupedReduceInvokable.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/GroupedReduceInvokable.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/GroupedReduceInvokable.java
new file mode 100755
index 0000000..c2177fa
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/GroupedReduceInvokable.java
@@ -0,0 +1,59 @@
+/*
+ * 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.invokable.operator;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.flink.api.common.functions.ReduceFunction;
+import org.apache.flink.api.java.functions.KeySelector;
+
+public class GroupedReduceInvokable<IN> extends StreamReduceInvokable<IN> {
+	private static final long serialVersionUID = 1L;
+
+	private KeySelector<IN, ?> keySelector;
+	private Map<Object, IN> values;
+	private IN reduced;
+
+	public GroupedReduceInvokable(ReduceFunction<IN> reducer, KeySelector<IN, ?> keySelector) {
+		super(reducer);
+		this.keySelector = keySelector;
+		values = new HashMap<Object, IN>();
+	}
+
+	@Override
+	protected void reduce() throws Exception {
+		Object key = nextRecord.getKey(keySelector);
+		currentValue = values.get(key);
+		nextValue = nextObject;
+		if (currentValue != null) {
+			callUserFunctionAndLogException();
+			values.put(key, reduced);
+			collector.collect(reduced);
+		} else {
+			values.put(key, nextValue);
+			collector.collect(nextValue);
+		}
+	}
+
+	@Override
+	protected void callUserFunction() throws Exception {
+		reduced = reducer.reduce(currentValue, nextValue);
+	}
+
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/f94112fb/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/GroupedWindowInvokable.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/GroupedWindowInvokable.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/GroupedWindowInvokable.java
new file mode 100644
index 0000000..997463c
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/GroupedWindowInvokable.java
@@ -0,0 +1,506 @@
+/*
+ * 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.invokable.operator;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.Map;
+import java.util.NoSuchElementException;
+
+import org.apache.flink.api.common.functions.Function;
+import org.apache.flink.api.common.functions.GroupReduceFunction;
+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.invokable.StreamInvokable;
+import org.apache.flink.streaming.api.streamrecord.StreamRecord;
+import org.apache.flink.streaming.api.windowing.policy.ActiveEvictionPolicy;
+import org.apache.flink.streaming.api.windowing.policy.ActiveTriggerCallback;
+import org.apache.flink.streaming.api.windowing.policy.ActiveTriggerPolicy;
+import org.apache.flink.streaming.api.windowing.policy.CloneableEvictionPolicy;
+import org.apache.flink.streaming.api.windowing.policy.CloneableTriggerPolicy;
+import org.apache.flink.streaming.api.windowing.policy.EvictionPolicy;
+import org.apache.flink.streaming.api.windowing.policy.TriggerPolicy;
+
+/**
+ * This invokable allows windowing based on {@link TriggerPolicy} and
+ * {@link EvictionPolicy} instances including their active and cloneable
+ * versions. It is additionally aware of the creation of windows per group.
+ * 
+ * A {@link KeySelector} is used to specify the key position or key extraction.
+ * The {@link ReduceFunction} will be executed on each group separately.
+ * Policies might either be centralized or distributed. It is not possible to
+ * use central and distributed eviction policies at the same time. A distributed
+ * policy have to be a {@link CloneableTriggerPolicy} or
+ * {@link CloneableEvictionPolicy} as it will be cloned to have separated
+ * instances for each group. At the startup time the distributed policies will
+ * be stored as sample, and only clones of them will be used to maintain the
+ * groups. Therefore, each group starts with the initial policy states.
+ * 
+ * While a distributed policy only gets notified with the elements belonging to
+ * the respective group, a centralized policy get notified with all arriving
+ * elements. When a centralized trigger occurred, all groups get triggered. This
+ * is done by submitting the element which caused the trigger as real element to
+ * the groups it belongs to and as fake element to all other groups. Within the
+ * groups the element might be further processed, causing more triggers,
+ * prenotifications of active distributed policies and evictions like usual.
+ * 
+ * Central policies can be instance of {@link ActiveTriggerPolicy} and also
+ * implement the
+ * {@link ActiveTriggerPolicy#createActiveTriggerRunnable(ActiveTriggerCallback)}
+ * method. Fake elements created on prenotification will be forwarded to all
+ * groups. The {@link ActiveTriggerCallback} is also implemented in a way, that
+ * it forwards/distributed calls all groups.
+ * 
+ * @param <IN>
+ *            The type of input elements handled by this operator invokable.
+ */
+public class GroupedWindowInvokable<IN, OUT> extends StreamInvokable<IN, OUT> {
+
+	/**
+	 * Auto-generated serial version UID
+	 */
+	private static final long serialVersionUID = -3469545957144404137L;
+
+	private KeySelector<IN, ?> keySelector;
+	private Configuration parameters;
+	private LinkedList<ActiveTriggerPolicy<IN>> activeCentralTriggerPolicies;
+	private LinkedList<TriggerPolicy<IN>> centralTriggerPolicies;
+	private LinkedList<ActiveEvictionPolicy<IN>> activeCentralEvictionPolicies;
+	private LinkedList<EvictionPolicy<IN>> centralEvictionPolicies;
+	private LinkedList<CloneableTriggerPolicy<IN>> distributedTriggerPolicies;
+	private LinkedList<CloneableEvictionPolicy<IN>> distributedEvictionPolicies;
+	private Map<Object, WindowInvokable<IN, OUT>> windowingGroups;
+	private LinkedList<Thread> activePolicyThreads;
+	private LinkedList<TriggerPolicy<IN>> currentTriggerPolicies;
+	private LinkedList<WindowInvokable<IN, OUT>> deleteOrderForCentralEviction;
+
+	/**
+	 * This constructor creates an instance of the grouped windowing invokable.
+	 * 
+	 * A {@link KeySelector} is used to specify the key position or key
+	 * extraction. The {@link ReduceFunction} will be executed on each group
+	 * separately. Policies might either be centralized or distributed. It is
+	 * not possible to use central and distributed eviction policies at the same
+	 * time. A distributed policy have to be a {@link CloneableTriggerPolicy} or
+	 * {@link CloneableEvictionPolicy} as it will be cloned to have separated
+	 * instances for each group. At the startup time the distributed policies
+	 * will be stored as sample, and only clones of them will be used to
+	 * maintain the groups. Therefore, each group starts with the initial policy
+	 * states.
+	 * 
+	 * While a distributed policy only gets notified with the elements belonging
+	 * to the respective group, a centralized policy get notified with all
+	 * arriving elements. When a centralized trigger occurred, all groups get
+	 * triggered. This is done by submitting the element which caused the
+	 * trigger as real element to the groups it belongs to and as fake element
+	 * to all other groups. Within the groups the element might be further
+	 * processed, causing more triggers, prenotifications of active distributed
+	 * policies and evictions like usual.
+	 * 
+	 * Central policies can be instance of {@link ActiveTriggerPolicy} and also
+	 * implement the
+	 * {@link ActiveTriggerPolicy#createActiveTriggerRunnable(ActiveTriggerCallback)}
+	 * method. Fake elements created on prenotification will be forwarded to all
+	 * groups. The {@link ActiveTriggerCallback} is also implemented in a way,
+	 * that it forwards/distributed calls all groups.
+	 * 
+	 * @param userFunction
+	 *            The user defined function.
+	 * @param keySelector
+	 *            A key selector to extract the key for the groups from the
+	 *            input data.
+	 * @param distributedTriggerPolicies
+	 *            Trigger policies to be distributed and maintained individually
+	 *            within each group.
+	 * @param distributedEvictionPolicies
+	 *            Eviction policies to be distributed and maintained
+	 *            individually within each group. Note that there cannot be
+	 *            both, central and distributed eviction policies at the same
+	 *            time.
+	 * @param centralTriggerPolicies
+	 *            Trigger policies which will only exist once at a central
+	 *            place. In case a central policy triggers, it will cause all
+	 *            groups to be emitted. (Remark: Empty groups cannot be emitted.
+	 *            If only one element is contained a group, this element itself
+	 *            is returned as aggregated result.)
+	 * @param centralEvictionPolicies
+	 *            Eviction which will only exist once at a central place. Note
+	 *            that there cannot be both, central and distributed eviction
+	 *            policies at the same time. The central eviction policy will
+	 *            work on an simulated element buffer containing all elements no
+	 *            matter which group they belong to.
+	 */
+	public GroupedWindowInvokable(Function userFunction, KeySelector<IN, ?> keySelector,
+			LinkedList<CloneableTriggerPolicy<IN>> distributedTriggerPolicies,
+			LinkedList<CloneableEvictionPolicy<IN>> distributedEvictionPolicies,
+			LinkedList<TriggerPolicy<IN>> centralTriggerPolicies,
+			LinkedList<EvictionPolicy<IN>> centralEvictionPolicies) {
+
+		super(userFunction);
+
+		this.keySelector = keySelector;
+
+		// handle the triggers
+		if (centralTriggerPolicies != null) {
+			this.centralTriggerPolicies = centralTriggerPolicies;
+			this.activeCentralTriggerPolicies = new LinkedList<ActiveTriggerPolicy<IN>>();
+
+			for (TriggerPolicy<IN> trigger : centralTriggerPolicies) {
+				if (trigger instanceof ActiveTriggerPolicy) {
+					this.activeCentralTriggerPolicies.add((ActiveTriggerPolicy<IN>) trigger);
+				}
+			}
+		} else {
+			this.centralTriggerPolicies = new LinkedList<TriggerPolicy<IN>>();
+		}
+
+		if (distributedTriggerPolicies != null) {
+			this.distributedTriggerPolicies = distributedTriggerPolicies;
+		} else {
+			this.distributedTriggerPolicies = new LinkedList<CloneableTriggerPolicy<IN>>();
+		}
+
+		if (distributedEvictionPolicies != null) {
+			this.distributedEvictionPolicies = distributedEvictionPolicies;
+		} else {
+			this.distributedEvictionPolicies = new LinkedList<CloneableEvictionPolicy<IN>>();
+		}
+
+		this.activeCentralEvictionPolicies = new LinkedList<ActiveEvictionPolicy<IN>>();
+
+		if (centralEvictionPolicies != null) {
+			this.centralEvictionPolicies = centralEvictionPolicies;
+
+			for (EvictionPolicy<IN> eviction : centralEvictionPolicies) {
+				if (eviction instanceof ActiveEvictionPolicy) {
+					this.activeCentralEvictionPolicies.add((ActiveEvictionPolicy<IN>) eviction);
+				}
+			}
+		} else {
+			this.centralEvictionPolicies = new LinkedList<EvictionPolicy<IN>>();
+		}
+
+		this.windowingGroups = new HashMap<Object, WindowInvokable<IN, OUT>>();
+		this.activePolicyThreads = new LinkedList<Thread>();
+		this.currentTriggerPolicies = new LinkedList<TriggerPolicy<IN>>();
+		this.deleteOrderForCentralEviction = new LinkedList<WindowInvokable<IN, OUT>>();
+
+		// check that not both, central and distributed eviction, is used at the
+		// same time.
+		if (!this.centralEvictionPolicies.isEmpty() && !this.distributedEvictionPolicies.isEmpty()) {
+			throw new UnsupportedOperationException(
+					"You can only use either central or distributed eviction policies but not both at the same time.");
+		}
+
+		// Check that there is at least one trigger and one eviction policy
+		if (this.centralEvictionPolicies.isEmpty() && this.distributedEvictionPolicies.isEmpty()) {
+			throw new UnsupportedOperationException(
+					"You have to define at least one eviction policy");
+		}
+		if (this.centralTriggerPolicies.isEmpty() && this.distributedTriggerPolicies.isEmpty()) {
+			throw new UnsupportedOperationException(
+					"You have to define at least one trigger policy");
+		}
+
+	}
+
+	@Override
+	public void invoke() throws Exception {
+		// Prevent empty data streams
+		if (readNext() == null) {
+			throw new RuntimeException("DataStream must not be empty");
+		}
+
+		// Continuously run
+		while (nextRecord != null) {
+			WindowInvokable<IN, OUT> groupInvokable = windowingGroups.get(keySelector
+					.getKey(nextRecord.getObject()));
+			if (groupInvokable == null) {
+				groupInvokable = makeNewGroup(nextRecord);
+			}
+
+			// Run the precalls for central active triggers
+			for (ActiveTriggerPolicy<IN> trigger : activeCentralTriggerPolicies) {
+				Object[] result = trigger.preNotifyTrigger(nextRecord.getObject());
+				for (Object in : result) {
+
+					// If central eviction is used, handle it here
+					if (!activeCentralEvictionPolicies.isEmpty()) {
+						evictElements(centralActiveEviction(in));
+					}
+
+					// process in groups
+					for (WindowInvokable<IN, OUT> group : windowingGroups.values()) {
+						group.processFakeElement(in, trigger);
+						checkForEmptyGroupBuffer(group);
+					}
+				}
+			}
+
+			// Process non-active central triggers
+			for (TriggerPolicy<IN> triggerPolicy : centralTriggerPolicies) {
+				if (triggerPolicy.notifyTrigger(nextRecord.getObject())) {
+					currentTriggerPolicies.add(triggerPolicy);
+				}
+			}
+
+			if (currentTriggerPolicies.isEmpty()) {
+
+				// only add the element to its group
+				groupInvokable.processRealElement(nextRecord.getObject());
+				checkForEmptyGroupBuffer(groupInvokable);
+
+				// If central eviction is used, handle it here
+				if (!centralEvictionPolicies.isEmpty()) {
+					evictElements(centralEviction(nextRecord.getObject(), false));
+					deleteOrderForCentralEviction.add(groupInvokable);
+				}
+
+			} else {
+
+				// call user function for all groups
+				for (WindowInvokable<IN, OUT> group : windowingGroups.values()) {
+					if (group == groupInvokable) {
+						// process real with initialized policies
+						group.processRealElement(nextRecord.getObject(), currentTriggerPolicies);
+					} else {
+						// process like a fake but also initialized with
+						// policies
+						group.externalTriggerFakeElement(nextRecord.getObject(),
+								currentTriggerPolicies);
+					}
+
+					// remove group in case it has an empty buffer
+					// checkForEmptyGroupBuffer(group);
+				}
+
+				// If central eviction is used, handle it here
+				if (!centralEvictionPolicies.isEmpty()) {
+					evictElements(centralEviction(nextRecord.getObject(), true));
+					deleteOrderForCentralEviction.add(groupInvokable);
+				}
+			}
+
+			// clear current trigger list
+			currentTriggerPolicies.clear();
+
+			// read next record
+			readNext();
+		}
+
+		// Stop all remaining threads from policies
+		for (Thread t : activePolicyThreads) {
+			t.interrupt();
+		}
+
+		// finally trigger the buffer.
+		for (WindowInvokable<IN, OUT> group : windowingGroups.values()) {
+			group.emitFinalWindow(centralTriggerPolicies);
+		}
+
+	}
+
+	/**
+	 * This method creates a new group. The method gets called in case an
+	 * element arrives which has a key which was not seen before. The method
+	 * created a nested {@link WindowInvokable} and therefore created clones of
+	 * all distributed trigger and eviction policies.
+	 * 
+	 * @param element
+	 *            The element which leads to the generation of a new group
+	 *            (previously unseen key)
+	 * @throws Exception
+	 *             In case the {@link KeySelector} throws an exception in
+	 *             {@link KeySelector#getKey(Object)}, the exception is not
+	 *             catched by this method.
+	 */
+	@SuppressWarnings("unchecked")
+	private WindowInvokable<IN, OUT> makeNewGroup(StreamRecord<IN> element) throws Exception {
+		// clone the policies
+		LinkedList<TriggerPolicy<IN>> clonedDistributedTriggerPolicies = new LinkedList<TriggerPolicy<IN>>();
+		LinkedList<EvictionPolicy<IN>> clonedDistributedEvictionPolicies = new LinkedList<EvictionPolicy<IN>>();
+		for (CloneableTriggerPolicy<IN> trigger : this.distributedTriggerPolicies) {
+			clonedDistributedTriggerPolicies.add(trigger.clone());
+		}
+		for (CloneableEvictionPolicy<IN> eviction : this.distributedEvictionPolicies) {
+			clonedDistributedEvictionPolicies.add(eviction.clone());
+		}
+
+		WindowInvokable<IN, OUT> groupInvokable;
+		if (userFunction instanceof ReduceFunction) {
+			groupInvokable = (WindowInvokable<IN, OUT>) new WindowReduceInvokable<IN>(
+					(ReduceFunction<IN>) userFunction, clonedDistributedTriggerPolicies,
+					clonedDistributedEvictionPolicies);
+		} else {
+			groupInvokable = new WindowGroupReduceInvokable<IN, OUT>(
+					(GroupReduceFunction<IN, OUT>) userFunction, clonedDistributedTriggerPolicies,
+					clonedDistributedEvictionPolicies);
+		}
+
+		groupInvokable.setup(taskContext);
+		groupInvokable.open(this.parameters);
+		windowingGroups.put(keySelector.getKey(element.getObject()), groupInvokable);
+
+		return groupInvokable;
+	}
+
+	@Override
+	public void open(Configuration parameters) throws Exception {
+		super.open(parameters);
+		this.parameters = parameters;
+		for (ActiveTriggerPolicy<IN> tp : activeCentralTriggerPolicies) {
+			Runnable target = tp.createActiveTriggerRunnable(new WindowingCallback(tp));
+			if (target != null) {
+				Thread thread = new Thread(target);
+				activePolicyThreads.add(thread);
+				thread.start();
+			}
+		}
+	};
+
+	/**
+	 * This method is used to notify central eviction policies with a real
+	 * element.
+	 * 
+	 * @param input
+	 *            the real element to notify the eviction policy.
+	 * @param triggered
+	 *            whether a central trigger occurred or not.
+	 * @return The number of elements to be deleted from the buffer.
+	 */
+	private int centralEviction(IN input, boolean triggered) {
+		// Process the evictions and take care of double evictions
+		// In case there are multiple eviction policies present,
+		// only the one with the highest return value is recognized.
+		int currentMaxEviction = 0;
+		for (EvictionPolicy<IN> evictionPolicy : centralEvictionPolicies) {
+			// use temporary variable to prevent multiple calls to
+			// notifyEviction
+			int tmp = evictionPolicy.notifyEviction(input, triggered,
+					deleteOrderForCentralEviction.size());
+			if (tmp > currentMaxEviction) {
+				currentMaxEviction = tmp;
+			}
+		}
+		return currentMaxEviction;
+	}
+
+	/**
+	 * This method is used to notify active central eviction policies with a
+	 * fake element.
+	 * 
+	 * @param input
+	 *            the fake element to notify the active central eviction
+	 *            policies.
+	 * @return The number of elements to be deleted from the buffer.
+	 */
+	private int centralActiveEviction(Object input) {
+		// Process the evictions and take care of double evictions
+		// In case there are multiple eviction policies present,
+		// only the one with the highest return value is recognized.
+		int currentMaxEviction = 0;
+		for (ActiveEvictionPolicy<IN> evictionPolicy : activeCentralEvictionPolicies) {
+			// use temporary variable to prevent multiple calls to
+			// notifyEviction
+			int tmp = evictionPolicy.notifyEvictionWithFakeElement(input,
+					deleteOrderForCentralEviction.size());
+			if (tmp > currentMaxEviction) {
+				currentMaxEviction = tmp;
+			}
+		}
+		return currentMaxEviction;
+	}
+
+	/**
+	 * This method is used in central eviction to delete a given number of
+	 * elements from the buffer.
+	 * 
+	 * @param numToEvict
+	 *            number of elements to delete from the virtual central element
+	 *            buffer.
+	 */
+	private void evictElements(int numToEvict) {
+		HashSet<WindowInvokable<IN, OUT>> usedGroups = new HashSet<WindowInvokable<IN, OUT>>();
+		for (; numToEvict > 0; numToEvict--) {
+			WindowInvokable<IN, OUT> currentGroup = deleteOrderForCentralEviction.getFirst();
+			// Do the eviction
+			currentGroup.evictFirst();
+			// Remember groups which possibly have an empty buffer after the
+			// eviction
+			usedGroups.add(currentGroup);
+			try {
+				deleteOrderForCentralEviction.removeFirst();
+			} catch (NoSuchElementException e) {
+				// when buffer is empty, ignore exception and stop deleting
+				break;
+			}
+
+		}
+
+		// Remove groups with empty buffer
+		for (WindowInvokable<IN, OUT> group : usedGroups) {
+			checkForEmptyGroupBuffer(group);
+		}
+	}
+
+	/**
+	 * Checks if the element buffer of a given windowing group is empty. If so,
+	 * the group will be deleted.
+	 * 
+	 * @param group
+	 *            The windowing group to be checked and and removed in case its
+	 *            buffer is empty.
+	 */
+	private void checkForEmptyGroupBuffer(WindowInvokable<IN, OUT> group) {
+		if (group.isBufferEmpty()) {
+			windowingGroups.remove(group);
+		}
+	}
+
+	/**
+	 * This callback class allows to handle the callbacks done by threads
+	 * defined in active trigger policies
+	 * 
+	 * @see ActiveTriggerPolicy#createActiveTriggerRunnable(ActiveTriggerCallback)
+	 */
+	private class WindowingCallback implements ActiveTriggerCallback {
+		private ActiveTriggerPolicy<IN> policy;
+
+		public WindowingCallback(ActiveTriggerPolicy<IN> policy) {
+			this.policy = policy;
+		}
+
+		@Override
+		public void sendFakeElement(Object datapoint) {
+
+			// If central eviction is used, handle it here
+			if (!centralEvictionPolicies.isEmpty()) {
+				evictElements(centralActiveEviction(datapoint));
+			}
+
+			// handle element in groups
+			for (WindowInvokable<IN, OUT> group : windowingGroups.values()) {
+				group.processFakeElement(datapoint, policy);
+				checkForEmptyGroupBuffer(group);
+			}
+		}
+
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/f94112fb/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/MapInvokable.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/MapInvokable.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/MapInvokable.java
new file mode 100644
index 0000000..7c8e577
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/MapInvokable.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.streaming.api.invokable.operator;
+
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.streaming.api.invokable.ChainableInvokable;
+
+public class MapInvokable<IN, OUT> extends ChainableInvokable<IN, OUT> {
+	private static final long serialVersionUID = 1L;
+
+	private MapFunction<IN, OUT> mapper;
+
+	public MapInvokable(MapFunction<IN, OUT> mapper) {
+		super(mapper);
+		this.mapper = mapper;
+	}
+
+	@Override
+	public void invoke() throws Exception {
+		while (readNext() != null) {
+			callUserFunctionAndLogException();
+		}
+	}
+
+	@Override
+	protected void callUserFunction() throws Exception {
+		collector.collect(mapper.map(nextObject));
+	}
+
+	@Override
+	public void collect(IN record) {
+		nextObject = copy(record);
+		callUserFunctionAndLogException();
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/f94112fb/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/ProjectInvokable.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/ProjectInvokable.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/ProjectInvokable.java
new file mode 100644
index 0000000..c9d9e5a
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/ProjectInvokable.java
@@ -0,0 +1,61 @@
+/**
+ * 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.invokable.operator;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.invokable.StreamInvokable;
+
+public class ProjectInvokable<IN, OUT extends Tuple> extends StreamInvokable<IN, OUT> {
+	private static final long serialVersionUID = 1L;
+
+	transient OUT outTuple;
+	TypeSerializer<OUT> outTypeSerializer;
+	int[] fields;
+	int numFields;
+
+	public ProjectInvokable(int[] fields, TypeInformation<OUT> outTypeInformation) {
+		super(null);
+		this.fields = fields;
+		this.numFields = this.fields.length;
+		this.outTypeSerializer = outTypeInformation.createSerializer();
+	}
+
+	@Override
+	public void invoke() throws Exception {
+		while (readNext() != null) {
+			callUserFunctionAndLogException();
+		}
+	}
+
+	@Override
+	protected void callUserFunction() throws Exception {
+		for (int i = 0; i < this.numFields; i++) {
+			outTuple.setField(nextRecord.getField(fields[i]), i);
+		}
+		collector.collect(outTuple);
+	}
+
+	@Override
+	public void open(Configuration config) throws Exception {
+		super.open(config);
+		outTuple = outTypeSerializer.createInstance();
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/f94112fb/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/StreamReduceInvokable.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/StreamReduceInvokable.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/StreamReduceInvokable.java
new file mode 100644
index 0000000..e1a56cc
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/StreamReduceInvokable.java
@@ -0,0 +1,69 @@
+/*
+ * 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.invokable.operator;
+
+import org.apache.flink.api.common.functions.ReduceFunction;
+import org.apache.flink.streaming.api.invokable.ChainableInvokable;
+
+public class StreamReduceInvokable<IN> extends ChainableInvokable<IN, IN> {
+	private static final long serialVersionUID = 1L;
+
+	protected ReduceFunction<IN> reducer;
+	protected IN currentValue;
+	protected IN nextValue;
+
+	public StreamReduceInvokable(ReduceFunction<IN> reducer) {
+		super(reducer);
+		this.reducer = reducer;
+		currentValue = null;
+	}
+
+	@Override
+	public void invoke() throws Exception {
+		while (readNext() != null) {
+			reduce();
+		}
+	}
+
+	protected void reduce() throws Exception {
+		callUserFunctionAndLogException();
+
+	}
+
+	@Override
+	protected void callUserFunction() throws Exception {
+
+		nextValue = nextObject;
+
+		if (currentValue != null) {
+			currentValue = reducer.reduce(currentValue, nextValue);
+		} else {
+			currentValue = nextValue;
+
+		}
+		collector.collect(currentValue);
+
+	}
+
+	@Override
+	public void collect(IN record) {
+		nextObject = copy(record);
+		callUserFunctionAndLogException();
+	}
+
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/f94112fb/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/WindowGroupReduceInvokable.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/WindowGroupReduceInvokable.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/WindowGroupReduceInvokable.java
new file mode 100644
index 0000000..b3fdfe8
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/WindowGroupReduceInvokable.java
@@ -0,0 +1,51 @@
+/*
+ * 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.invokable.operator;
+
+import java.util.LinkedList;
+
+import org.apache.flink.api.common.functions.GroupReduceFunction;
+import org.apache.flink.streaming.api.windowing.policy.EvictionPolicy;
+import org.apache.flink.streaming.api.windowing.policy.TriggerPolicy;
+
+public class WindowGroupReduceInvokable<IN, OUT> extends WindowInvokable<IN, OUT> {
+
+	private static final long serialVersionUID = 1L;
+	GroupReduceFunction<IN, OUT> reducer;
+
+	public WindowGroupReduceInvokable(GroupReduceFunction<IN, OUT> userFunction,
+			LinkedList<TriggerPolicy<IN>> triggerPolicies,
+			LinkedList<EvictionPolicy<IN>> evictionPolicies) {
+		super(userFunction, triggerPolicies, evictionPolicies);
+		this.reducer = userFunction;
+	}
+
+	@Override
+	protected void callUserFunction() throws Exception {
+		reducer.reduce(copyBuffer(), collector);
+	}
+
+	public LinkedList<IN> copyBuffer() {
+		LinkedList<IN> copy = new LinkedList<IN>();
+		for (IN element : buffer) {
+			copy.add(copy(element));
+		}
+		return copy;
+	}
+
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/f94112fb/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/WindowInvokable.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/WindowInvokable.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/WindowInvokable.java
new file mode 100644
index 0000000..ea891c9
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/WindowInvokable.java
@@ -0,0 +1,382 @@
+/*
+ * 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.invokable.operator;
+
+import java.util.LinkedList;
+import java.util.List;
+import java.util.NoSuchElementException;
+
+import org.apache.flink.api.common.functions.Function;
+import org.apache.flink.api.common.functions.ReduceFunction;
+import org.apache.flink.streaming.api.invokable.StreamInvokable;
+import org.apache.flink.streaming.api.windowing.policy.ActiveEvictionPolicy;
+import org.apache.flink.streaming.api.windowing.policy.ActiveTriggerCallback;
+import org.apache.flink.streaming.api.windowing.policy.ActiveTriggerPolicy;
+import org.apache.flink.streaming.api.windowing.policy.EvictionPolicy;
+import org.apache.flink.streaming.api.windowing.policy.TriggerPolicy;
+
+public abstract class WindowInvokable<IN, OUT> extends StreamInvokable<IN, OUT> {
+
+	/**
+	 * Auto-generated serial version UID
+	 */
+	private static final long serialVersionUID = -8038984294071650730L;
+
+	private LinkedList<TriggerPolicy<IN>> triggerPolicies;
+	private LinkedList<EvictionPolicy<IN>> evictionPolicies;
+	private LinkedList<ActiveTriggerPolicy<IN>> activeTriggerPolicies;
+	private LinkedList<ActiveEvictionPolicy<IN>> activeEvictionPolicies;
+	private LinkedList<Thread> activePolicyTreads;
+	protected LinkedList<IN> buffer;
+	private LinkedList<TriggerPolicy<IN>> currentTriggerPolicies;
+
+	/**
+	 * This constructor created a windowing invokable using trigger and eviction
+	 * policies.
+	 * 
+	 * @param userFunction
+	 *            The user defined {@link ReduceFunction}
+	 * @param triggerPolicies
+	 *            A list of {@link TriggerPolicy}s and/or
+	 *            {@link ActiveTriggerPolicy}s
+	 * @param evictionPolicies
+	 *            A list of {@link EvictionPolicy}s and/or
+	 *            {@link ActiveEvictionPolicy}s
+	 */
+	public WindowInvokable(Function userFunction, LinkedList<TriggerPolicy<IN>> triggerPolicies,
+			LinkedList<EvictionPolicy<IN>> evictionPolicies) {
+		super(userFunction);
+
+		this.triggerPolicies = triggerPolicies;
+		this.evictionPolicies = evictionPolicies;
+
+		activeTriggerPolicies = new LinkedList<ActiveTriggerPolicy<IN>>();
+		for (TriggerPolicy<IN> tp : triggerPolicies) {
+			if (tp instanceof ActiveTriggerPolicy) {
+				activeTriggerPolicies.add((ActiveTriggerPolicy<IN>) tp);
+			}
+		}
+
+		activeEvictionPolicies = new LinkedList<ActiveEvictionPolicy<IN>>();
+		for (EvictionPolicy<IN> ep : evictionPolicies) {
+			if (ep instanceof ActiveEvictionPolicy) {
+				activeEvictionPolicies.add((ActiveEvictionPolicy<IN>) ep);
+			}
+		}
+
+		this.activePolicyTreads = new LinkedList<Thread>();
+		this.buffer = new LinkedList<IN>();
+		this.currentTriggerPolicies = new LinkedList<TriggerPolicy<IN>>();
+	}
+
+	@Override
+	public void open(org.apache.flink.configuration.Configuration parameters) throws Exception {
+		super.open(parameters);
+		for (ActiveTriggerPolicy<IN> tp : activeTriggerPolicies) {
+			Runnable target = tp.createActiveTriggerRunnable(new WindowingCallback(tp));
+			if (target != null) {
+				Thread thread = new Thread(target);
+				activePolicyTreads.add(thread);
+				thread.start();
+			}
+		}
+	};
+
+	/**
+	 * This class allows the active trigger threads to call back and push fake
+	 * elements at any time.
+	 */
+	private class WindowingCallback implements ActiveTriggerCallback {
+		private ActiveTriggerPolicy<IN> policy;
+
+		public WindowingCallback(ActiveTriggerPolicy<IN> policy) {
+			this.policy = policy;
+		}
+
+		@Override
+		public void sendFakeElement(Object datapoint) {
+			processFakeElement(datapoint, this.policy);
+		}
+
+	}
+
+	@Override
+	public void invoke() throws Exception {
+
+		// Prevent empty data streams
+		if (readNext() == null) {
+			throw new RuntimeException("DataStream must not be empty");
+		}
+
+		// Continuously run
+		while (nextRecord != null) {
+			processRealElement(nextRecord.getObject());
+
+			// Load next StreamRecord
+			readNext();
+		}
+
+		// Stop all remaining threads from policies
+		for (Thread t : activePolicyTreads) {
+			t.interrupt();
+		}
+
+		// finally trigger the buffer.
+		emitFinalWindow(null);
+
+	}
+
+	/**
+	 * This method gets called in case of an grouped windowing in case central
+	 * trigger occurred and the arriving element causing the trigger is not part
+	 * of this group.
+	 * 
+	 * Remark: This is NOT the same as
+	 * {@link WindowInvokable#processFakeElement(Object, TriggerPolicy)}! Here
+	 * the eviction using active policies takes place after the call to the UDF.
+	 * Usually it is done before when fake elements get submitted. This special
+	 * behaviour is needed to allow the {@link GroupedWindowInvokable} to send
+	 * central triggers to all groups, even if the current element does not
+	 * belong to the group.
+	 * 
+	 * @param input
+	 *            a fake input element
+	 * @param policies
+	 *            the list of policies which caused the call with this fake
+	 *            element
+	 */
+	protected synchronized void externalTriggerFakeElement(IN input,
+			List<TriggerPolicy<IN>> policies) {
+
+		// Set the current triggers
+		currentTriggerPolicies.addAll(policies);
+
+		// emit
+		callUserFunctionAndLogException();
+
+		// clear the flag collection
+		currentTriggerPolicies.clear();
+
+		// Process the evictions and take care of double evictions
+		// In case there are multiple eviction policies present,
+		// only the one with the highest return value is recognized.
+		int currentMaxEviction = 0;
+		for (ActiveEvictionPolicy<IN> evictionPolicy : activeEvictionPolicies) {
+			// use temporary variable to prevent multiple calls to
+			// notifyEviction
+			int tmp = evictionPolicy.notifyEvictionWithFakeElement(input, buffer.size());
+			if (tmp > currentMaxEviction) {
+				currentMaxEviction = tmp;
+			}
+		}
+
+		for (int i = 0; i < currentMaxEviction; i++) {
+			try {
+				buffer.removeFirst();
+			} catch (NoSuchElementException e) {
+				// In case no more elements are in the buffer:
+				// Prevent failure and stop deleting.
+				break;
+			}
+		}
+	}
+
+	/**
+	 * This method processed an arrived fake element The method is synchronized
+	 * to ensure that it cannot interleave with
+	 * {@link WindowInvokable#processRealElement(Object)}
+	 * 
+	 * @param input
+	 *            a fake input element
+	 * @param currentPolicy
+	 *            the policy which produced this fake element
+	 */
+	protected synchronized void processFakeElement(Object input, TriggerPolicy<IN> currentPolicy) {
+
+		// Process the evictions and take care of double evictions
+		// In case there are multiple eviction policies present,
+		// only the one with the highest return value is recognized.
+		int currentMaxEviction = 0;
+		for (ActiveEvictionPolicy<IN> evictionPolicy : activeEvictionPolicies) {
+			// use temporary variable to prevent multiple calls to
+			// notifyEviction
+			int tmp = evictionPolicy.notifyEvictionWithFakeElement(input, buffer.size());
+			if (tmp > currentMaxEviction) {
+				currentMaxEviction = tmp;
+			}
+		}
+
+		for (int i = 0; i < currentMaxEviction; i++) {
+			try {
+				buffer.removeFirst();
+			} catch (NoSuchElementException e) {
+				// In case no more elements are in the buffer:
+				// Prevent failure and stop deleting.
+				break;
+			}
+		}
+
+		// Set the current trigger
+		currentTriggerPolicies.add(currentPolicy);
+
+		// emit
+		callUserFunctionAndLogException();
+
+		// clear the flag collection
+		currentTriggerPolicies.clear();
+	}
+
+	/**
+	 * This method processed an arrived real element The method is synchronized
+	 * to ensure that it cannot interleave with
+	 * {@link WindowInvokable#processFakeElement(Object)}.
+	 * 
+	 * @param input
+	 *            a real input element
+	 * @param triggerPolicies
+	 *            Allows to set trigger policies which are maintained
+	 *            externally. This is the case for central policies in
+	 *            {@link GroupedWindowInvokable}.
+	 */
+	protected synchronized void processRealElement(IN input, List<TriggerPolicy<IN>> triggerPolicies) {
+		this.currentTriggerPolicies.addAll(triggerPolicies);
+		processRealElement(input);
+	}
+
+	/**
+	 * This method processed an arrived real element The method is synchronized
+	 * to ensure that it cannot interleave with
+	 * {@link WindowInvokable#processFakeElement(Object)}
+	 * 
+	 * @param input
+	 *            a real input element
+	 */
+	protected synchronized void processRealElement(IN input) {
+
+		// Run the precalls to detect missed windows
+		for (ActiveTriggerPolicy<IN> trigger : activeTriggerPolicies) {
+			// Remark: In case multiple active triggers are present the ordering
+			// of the different fake elements returned by this triggers becomes
+			// a problem. This might lead to unexpected results...
+			// Should we limit the number of active triggers to 0 or 1?
+			Object[] result = trigger.preNotifyTrigger(input);
+			for (Object in : result) {
+				processFakeElement(in, trigger);
+			}
+		}
+
+		// Remember if a trigger occurred
+		boolean isTriggered = false;
+
+		// Process the triggers
+		for (TriggerPolicy<IN> triggerPolicy : triggerPolicies) {
+			if (triggerPolicy.notifyTrigger(input)) {
+				currentTriggerPolicies.add(triggerPolicy);
+			}
+		}
+
+		// call user function
+		if (!currentTriggerPolicies.isEmpty()) {
+			// emit
+			callUserFunctionAndLogException();
+
+			// clear the flag collection
+			currentTriggerPolicies.clear();
+
+			// remember trigger
+			isTriggered = true;
+		}
+
+		// Process the evictions and take care of double evictions
+		// In case there are multiple eviction policies present,
+		// only the one with the highest return value is recognized.
+		int currentMaxEviction = 0;
+
+		for (EvictionPolicy<IN> evictionPolicy : evictionPolicies) {
+			// use temporary variable to prevent multiple calls to
+			// notifyEviction
+			int tmp = evictionPolicy.notifyEviction(input, isTriggered, buffer.size());
+			if (tmp > currentMaxEviction) {
+				currentMaxEviction = tmp;
+			}
+		}
+
+		for (int i = 0; i < currentMaxEviction; i++) {
+			try {
+				buffer.removeFirst();
+			} catch (NoSuchElementException e) {
+				// In case no more elements are in the buffer:
+				// Prevent failure and stop deleting.
+				break;
+			}
+		}
+
+		// Add the current element to the buffer
+		buffer.add(input);
+
+	}
+
+	/**
+	 * This method removes the first element from the element buffer. It is used
+	 * to provide central evictions in {@link GroupedWindowInvokable}
+	 */
+	protected synchronized void evictFirst() {
+		try {
+			buffer.removeFirst();
+		} catch (NoSuchElementException e) {
+			// ignore exception
+		}
+	}
+
+	/**
+	 * This method returns whether the element buffer is empty or not. It is
+	 * used to figure out if a group can be deleted or not when
+	 * {@link GroupedWindowInvokable} is used.
+	 * 
+	 * @return true in case the buffer is empty otherwise false.
+	 */
+	protected boolean isBufferEmpty() {
+		return buffer.isEmpty();
+	}
+
+	/**
+	 * This method does the final reduce at the end of the stream and emits the
+	 * result.
+	 * 
+	 * @param centralTriggerPolicies
+	 *            Allows to set trigger policies which are maintained
+	 *            externally. This is the case for central policies in
+	 *            {@link GroupedWindowInvokable}.
+	 */
+	protected void emitFinalWindow(List<TriggerPolicy<IN>> centralTriggerPolicies) {
+		if (!buffer.isEmpty()) {
+			currentTriggerPolicies.clear();
+
+			if (centralTriggerPolicies != null) {
+				currentTriggerPolicies.addAll(centralTriggerPolicies);
+			}
+
+			for (TriggerPolicy<IN> policy : triggerPolicies) {
+				currentTriggerPolicies.add(policy);
+			}
+
+			callUserFunctionAndLogException();
+		}
+	}
+
+}


Mime
View raw message