flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From aljos...@apache.org
Subject [2/5] flink git commit: [FLINK-4460] Make CoProcessFunction abstract, add default onTime() method
Date Mon, 06 Mar 2017 15:52:10 GMT
[FLINK-4460] Make CoProcessFunction abstract, add default onTime() method

This is in preparation of allowing CoProcessFunction on a non-keyed
connected stream.  we will use it to allow side outputs from the
ProcessFunction Context.


Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/e12f3203
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/e12f3203
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/e12f3203

Branch: refs/heads/master
Commit: e12f3203eda7f071daeea7c59ca32b3db7763b68
Parents: 0228676
Author: Aljoscha Krettek <aljoscha.krettek@gmail.com>
Authored: Wed Mar 1 11:33:03 2017 +0100
Committer: Aljoscha Krettek <aljoscha.krettek@gmail.com>
Committed: Mon Mar 6 12:26:16 2017 +0100

----------------------------------------------------------------------
 .../api/datastream/ConnectedStreams.java        |  9 -----
 .../api/functions/co/CoProcessFunction.java     | 24 ++++++------
 .../api/functions/co/RichCoProcessFunction.java | 41 --------------------
 .../api/operators/co/CoProcessOperator.java     | 23 +++++------
 .../api/operators/co/CoProcessOperatorTest.java | 15 ++++---
 .../streaming/api/scala/ConnectedStreams.scala  |  2 +-
 6 files changed, 32 insertions(+), 82 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/e12f3203/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/ConnectedStreams.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/ConnectedStreams.java
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/ConnectedStreams.java
index 96a08d3..66601a7 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/ConnectedStreams.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/ConnectedStreams.java
@@ -28,7 +28,6 @@ import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
 import org.apache.flink.streaming.api.functions.co.CoMapFunction;
 import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
-import org.apache.flink.streaming.api.functions.co.RichCoProcessFunction;
 import org.apache.flink.streaming.api.operators.TwoInputStreamOperator;
 import org.apache.flink.streaming.api.operators.co.CoStreamFlatMap;
 import org.apache.flink.streaming.api.operators.co.CoStreamMap;
@@ -243,10 +242,6 @@ public class ConnectedStreams<IN1, IN2> {
 	 * function can also query the time and set timers. When reacting to the firing of set timers
 	 * the function can directly emit elements and/or register yet more timers.
 	 *
-	 * <p>A {@link RichCoProcessFunction}
-	 * can be used to gain access to features provided by the
-	 * {@link org.apache.flink.api.common.functions.RichFunction} interface.
-	 *
 	 * @param coProcessFunction The {@link CoProcessFunction} that is called for each element
 	 *                      in the stream.
 	 *
@@ -274,10 +269,6 @@ public class ConnectedStreams<IN1, IN2> {
 	 * this function can also query the time and set timers. When reacting to the firing of
set
 	 * timers the function can directly emit elements and/or register yet more timers.
 	 *
-	 * <p>A {@link RichCoProcessFunction}
-	 * can be used to gain access to features provided by the
-	 * {@link org.apache.flink.api.common.functions.RichFunction} interface.
-	 *
 	 * @param coProcessFunction The {@link CoProcessFunction} that is called for each element
 	 *                      in the stream.
 	 *

http://git-wip-us.apache.org/repos/asf/flink/blob/e12f3203/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/co/CoProcessFunction.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/co/CoProcessFunction.java
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/co/CoProcessFunction.java
index feff8fb..8811e32 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/co/CoProcessFunction.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/co/CoProcessFunction.java
@@ -19,13 +19,11 @@
 package org.apache.flink.streaming.api.functions.co;
 
 import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.api.common.functions.Function;
+import org.apache.flink.api.common.functions.AbstractRichFunction;
 import org.apache.flink.streaming.api.TimeDomain;
 import org.apache.flink.streaming.api.TimerService;
 import org.apache.flink.util.Collector;
 
-import java.io.Serializable;
-
 /**
  * A function that processes elements of two streams and produces a single output one.
  *
@@ -46,7 +44,9 @@ import java.io.Serializable;
  * @param <OUT> Output type.
  */
 @PublicEvolving
-public interface CoProcessFunction<IN1, IN2, OUT> extends Function, Serializable {
+public abstract class CoProcessFunction<IN1, IN2, OUT> extends AbstractRichFunction
{
+
+	private static final long serialVersionUID = 1L;
 
 	/**
 	 * This method is called for each element in the first of the connected streams.
@@ -63,7 +63,7 @@ public interface CoProcessFunction<IN1, IN2, OUT> extends Function,
Serializable
 	 * @throws Exception The function may throw exceptions which cause the streaming program
 	 *                   to fail and go into recovery.
 	 */
-	void processElement1(IN1 value, Context ctx, Collector<OUT> out) throws Exception;
+	public abstract void processElement1(IN1 value, Context ctx, Collector<OUT> out) throws
Exception;
 
 	/**
 	 * This method is called for each element in the second of the connected streams.
@@ -80,7 +80,7 @@ public interface CoProcessFunction<IN1, IN2, OUT> extends Function,
Serializable
 	 * @throws Exception The function may throw exceptions which cause the streaming program
 	 *                   to fail and go into recovery.
 	 */
-	void processElement2(IN2 value, Context ctx, Collector<OUT> out) throws Exception;
+	public abstract void processElement2(IN2 value, Context ctx, Collector<OUT> out) throws
Exception;
 
 	/**
 	 * Called when a timer set using {@link TimerService} fires.
@@ -95,14 +95,14 @@ public interface CoProcessFunction<IN1, IN2, OUT> extends Function,
Serializable
 	 * @throws Exception This method may throw exceptions. Throwing an exception will cause
the operation
 	 *                   to fail and may trigger recovery.
 	 */
-	void onTimer(long timestamp, OnTimerContext ctx, Collector<OUT> out) throws Exception
;
+	public void onTimer(long timestamp, OnTimerContext ctx, Collector<OUT> out) throws
Exception {}
 
 	/**
 	 * Information available in an invocation of {@link #processElement1(Object, Context, Collector)}/
 	 * {@link #processElement2(Object, Context, Collector)}
 	 * or {@link #onTimer(long, OnTimerContext, Collector)}.
 	 */
-	interface Context {
+	public abstract class Context {
 
 		/**
 		 * Timestamp of the element currently being processed or timestamp of a firing timer.
@@ -110,21 +110,21 @@ public interface CoProcessFunction<IN1, IN2, OUT> extends Function,
Serializable
 		 * <p>This might be {@code null}, for example if the time characteristic of your
program
 		 * is set to {@link org.apache.flink.streaming.api.TimeCharacteristic#ProcessingTime}.
 		 */
-		Long timestamp();
+		public abstract Long timestamp();
 
 		/**
 		 * A {@link TimerService} for querying time and registering timers.
 		 */
-		TimerService timerService();
+		public abstract TimerService timerService();
 	}
 
 	/**
 	 * Information available in an invocation of {@link #onTimer(long, OnTimerContext, Collector)}.
 	 */
-	interface OnTimerContext extends Context {
+	public abstract class OnTimerContext extends Context {
 		/**
 		 * The {@link TimeDomain} of the firing timer.
 		 */
-		TimeDomain timeDomain();
+		public abstract TimeDomain timeDomain();
 	}
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/e12f3203/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/co/RichCoProcessFunction.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/co/RichCoProcessFunction.java
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/co/RichCoProcessFunction.java
deleted file mode 100644
index 0fcea91..0000000
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/co/RichCoProcessFunction.java
+++ /dev/null
@@ -1,41 +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.streaming.api.functions.co;
-
-import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.api.common.functions.AbstractRichFunction;
-import org.apache.flink.api.common.functions.RichFunction;
-
-/**
- * Rich variant of the {@link CoProcessFunction}. As a {@link RichFunction}, it gives
- * access to the {@link org.apache.flink.api.common.functions.RuntimeContext} and provides
- * setup and teardown methods: {@link RichFunction#open(org.apache.flink.configuration.Configuration)}
- * and {@link RichFunction#close()}.
- *
- * @param <IN1> Type of the first input.
- * @param <IN2> Type of the second input.
- * @param <OUT> Type of the returned elements.
- */
-@PublicEvolving
-public abstract class RichCoProcessFunction<IN1, IN2, OUT>
-		extends AbstractRichFunction
-		implements CoProcessFunction<IN1, IN2, OUT> {
-
-	private static final long serialVersionUID = 1L;
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/e12f3203/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/co/CoProcessOperator.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/co/CoProcessOperator.java
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/co/CoProcessOperator.java
index e6c3d3f..ed99815 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/co/CoProcessOperator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/co/CoProcessOperator.java
@@ -44,11 +44,9 @@ public class CoProcessOperator<K, IN1, IN2, OUT>
 
 	private transient TimestampedCollector<OUT> collector;
 
-	private transient TimerService timerService;
+	private transient ContextImpl<IN1, IN2, OUT> context;
 
-	private transient ContextImpl context;
-
-	private transient OnTimerContextImpl onTimerContext;
+	private transient OnTimerContextImpl<IN1, IN2, OUT> onTimerContext;
 
 	public CoProcessOperator(CoProcessFunction<IN1, IN2, OUT> flatMapper) {
 		super(flatMapper);
@@ -62,10 +60,10 @@ public class CoProcessOperator<K, IN1, IN2, OUT>
 		InternalTimerService<VoidNamespace> internalTimerService =
 				getInternalTimerService("user-timers", VoidNamespaceSerializer.INSTANCE, this);
 
-		this.timerService = new SimpleTimerService(internalTimerService);
+		TimerService timerService = new SimpleTimerService(internalTimerService);
 
-		context = new ContextImpl(timerService);
-		onTimerContext = new OnTimerContextImpl(timerService);
+		context = new ContextImpl<>(userFunction, timerService);
+		onTimerContext = new OnTimerContextImpl<>(userFunction, timerService);
 	}
 
 	@Override
@@ -108,13 +106,14 @@ public class CoProcessOperator<K, IN1, IN2, OUT>
 		return collector;
 	}
 
-	private static class ContextImpl implements CoProcessFunction.Context {
+	private static class ContextImpl<IN1, IN2, OUT> extends CoProcessFunction<IN1,
IN2, OUT>.Context {
 
 		private final TimerService timerService;
 
 		private StreamRecord<?> element;
 
-		ContextImpl(TimerService timerService) {
+		ContextImpl(CoProcessFunction<IN1, IN2, OUT> function, TimerService timerService)
{
+			function.super();
 			this.timerService = checkNotNull(timerService);
 		}
 
@@ -135,7 +134,8 @@ public class CoProcessOperator<K, IN1, IN2, OUT>
 		}
 	}
 
-	private static class OnTimerContextImpl implements CoProcessFunction.OnTimerContext {
+	private static class OnTimerContextImpl<IN1, IN2, OUT>
+			extends CoProcessFunction<IN1, IN2, OUT>.OnTimerContext {
 
 		private final TimerService timerService;
 
@@ -143,7 +143,8 @@ public class CoProcessOperator<K, IN1, IN2, OUT>
 
 		private InternalTimer<?, VoidNamespace> timer;
 
-		OnTimerContextImpl(TimerService timerService) {
+		OnTimerContextImpl(CoProcessFunction<IN1, IN2, OUT> function, TimerService timerService)
{
+			function.super();
 			this.timerService = checkNotNull(timerService);
 		}
 

http://git-wip-us.apache.org/repos/asf/flink/blob/e12f3203/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/co/CoProcessOperatorTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/co/CoProcessOperatorTest.java
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/co/CoProcessOperatorTest.java
index eea428f..9d7c444 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/co/CoProcessOperatorTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/co/CoProcessOperatorTest.java
@@ -23,7 +23,6 @@ import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
 import org.apache.flink.api.common.typeutils.base.StringSerializer;
 import org.apache.flink.api.java.functions.KeySelector;
 import org.apache.flink.streaming.api.TimeDomain;
-import org.apache.flink.streaming.api.functions.co.RichCoProcessFunction;
 import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
 import org.apache.flink.streaming.api.watermark.Watermark;
 import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
@@ -344,7 +343,7 @@ public class CoProcessOperatorTest extends TestLogger {
 		}
 	}
 
-	private static class WatermarkQueryingProcessFunction implements CoProcessFunction<Integer,
String, String> {
+	private static class WatermarkQueryingProcessFunction extends CoProcessFunction<Integer,
String, String> {
 
 		private static final long serialVersionUID = 1L;
 
@@ -366,7 +365,7 @@ public class CoProcessOperatorTest extends TestLogger {
 		}
 	}
 
-	private static class EventTimeTriggeringProcessFunction implements CoProcessFunction<Integer,
String, String> {
+	private static class EventTimeTriggeringProcessFunction extends CoProcessFunction<Integer,
String, String> {
 
 		private static final long serialVersionUID = 1L;
 
@@ -393,7 +392,7 @@ public class CoProcessOperatorTest extends TestLogger {
 		}
 	}
 
-	private static class EventTimeTriggeringStatefulProcessFunction extends RichCoProcessFunction<Integer,
String, String> {
+	private static class EventTimeTriggeringStatefulProcessFunction extends CoProcessFunction<Integer,
String, String> {
 
 		private static final long serialVersionUID = 1L;
 
@@ -425,7 +424,7 @@ public class CoProcessOperatorTest extends TestLogger {
 		}
 	}
 
-	private static class ProcessingTimeTriggeringProcessFunction implements CoProcessFunction<Integer,
String, String> {
+	private static class ProcessingTimeTriggeringProcessFunction extends CoProcessFunction<Integer,
String, String> {
 
 		private static final long serialVersionUID = 1L;
 
@@ -452,7 +451,7 @@ public class CoProcessOperatorTest extends TestLogger {
 		}
 	}
 
-	private static class ProcessingTimeQueryingProcessFunction implements CoProcessFunction<Integer,
String, String> {
+	private static class ProcessingTimeQueryingProcessFunction extends CoProcessFunction<Integer,
String, String> {
 
 		private static final long serialVersionUID = 1L;
 
@@ -474,7 +473,7 @@ public class CoProcessOperatorTest extends TestLogger {
 		}
 	}
 
-	private static class ProcessingTimeTriggeringStatefulProcessFunction extends RichCoProcessFunction<Integer,
String, String> {
+	private static class ProcessingTimeTriggeringStatefulProcessFunction extends CoProcessFunction<Integer,
String, String> {
 
 		private static final long serialVersionUID = 1L;
 
@@ -506,7 +505,7 @@ public class CoProcessOperatorTest extends TestLogger {
 		}
 	}
 
-	private static class BothTriggeringProcessFunction implements CoProcessFunction<Integer,
String, String> {
+	private static class BothTriggeringProcessFunction extends CoProcessFunction<Integer,
String, String> {
 
 		private static final long serialVersionUID = 1L;
 

http://git-wip-us.apache.org/repos/asf/flink/blob/e12f3203/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/ConnectedStreams.scala
----------------------------------------------------------------------
diff --git a/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/ConnectedStreams.scala
b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/ConnectedStreams.scala
index a7325a4..cebda5d 100644
--- a/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/ConnectedStreams.scala
+++ b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/ConnectedStreams.scala
@@ -23,7 +23,7 @@ import org.apache.flink.api.common.typeinfo.TypeInformation
 import org.apache.flink.api.java.functions.KeySelector
 import org.apache.flink.api.java.typeutils.ResultTypeQueryable
 import org.apache.flink.streaming.api.datastream.{ConnectedStreams => JavaCStream, DataStream
=> JavaStream}
-import org.apache.flink.streaming.api.functions.co.{CoFlatMapFunction, CoMapFunction, CoProcessFunction,
RichCoProcessFunction}
+import org.apache.flink.streaming.api.functions.co.{CoFlatMapFunction, CoMapFunction, CoProcessFunction}
 import org.apache.flink.streaming.api.operators.TwoInputStreamOperator
 import org.apache.flink.util.Collector
 


Mime
View raw message