From 09f26b3647f3924f3bf235f1c6ca38fdf1dfeaca Mon Sep 17 00:00:00 2001
From: Alexander Pivovarov <apivovarov@gmail.com>
Date: Fri, 9 Sep 2016 22:24:43 -0700
Subject: [PATCH] [FLINK-4609] Remove redundant check for null in CrossOperator

---
 .../api/java/operators/CrossOperator.java     | 23 ++++++-------------
 .../api/java/operators/TwoInputOperator.java  |  5 ++--
 2 files changed, 10 insertions(+), 18 deletions(-)

diff --git a/flink-java/src/main/java/org/apache/flink/api/java/operators/CrossOperator.java b/flink-java/src/main/java/org/apache/flink/api/java/operators/CrossOperator.java
index 36e6c1c2a0916..3fdc51dbfde70 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/operators/CrossOperator.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/operators/CrossOperator.java
@@ -20,9 +20,9 @@
 
 import java.util.Arrays;
 
-import org.apache.flink.annotation.PublicEvolving;
 import org.apache.flink.annotation.Internal;
 import org.apache.flink.annotation.Public;
+import org.apache.flink.annotation.PublicEvolving;
 import org.apache.flink.api.common.InvalidProgramException;
 import org.apache.flink.api.common.functions.CrossFunction;
 import org.apache.flink.api.common.operators.BinaryOperatorInformation;
@@ -124,21 +124,12 @@ protected org.apache.flink.api.common.operators.base.CrossOperatorBase<I1, I2, O
 	@Public
 	public static final class DefaultCross<I1, I2> extends CrossOperator<I1, I2, Tuple2<I1, I2>>  {
 
-		private final DataSet<I1> input1;
-		private final DataSet<I2> input2;
-
 		public DefaultCross(DataSet<I1> input1, DataSet<I2> input2, CrossHint hint, String defaultName) {
-			
 			super(input1, input2, new DefaultCrossFunction<I1, I2>(),
-					new TupleTypeInfo<Tuple2<I1, I2>>(input1.getType(), input2.getType()),
-					hint, defaultName);
-
-			if (input1 == null || input2 == null) {
-				throw new NullPointerException();
-			}
-
-			this.input1 = input1;
-			this.input2 = input2;
+				new TupleTypeInfo<Tuple2<I1, I2>>(
+					Preconditions.checkNotNull(input1, "input1 is null").getType(),
+					Preconditions.checkNotNull(input2, "input2 is null").getType()),
+				hint, defaultName);
 		}
 
 		/**
@@ -155,9 +146,9 @@ public <R> CrossOperator<I1, I2, R> with(CrossFunction<I1, I2, R> function) {
 			if (function == null) {
 				throw new NullPointerException("Cross function must not be null.");
 			}
-			TypeInformation<R> returnType = TypeExtractor.getCrossReturnTypes(function, input1.getType(), input2.getType(),
+			TypeInformation<R> returnType = TypeExtractor.getCrossReturnTypes(function, getInput1().getType(), getInput2().getType(),
 					super.getDefaultName(), true);
-			return new CrossOperator<I1, I2, R>(input1, input2, clean(function), returnType, 
+			return new CrossOperator<I1, I2, R>(getInput1(), getInput2(), clean(function), returnType,
 					getCrossHint(), Utils.getCallLocationName());
 		}
 		
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/operators/TwoInputOperator.java b/flink-java/src/main/java/org/apache/flink/api/java/operators/TwoInputOperator.java
index c64882d37a699..28dec3261efeb 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/operators/TwoInputOperator.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/operators/TwoInputOperator.java
@@ -21,6 +21,7 @@
 import org.apache.flink.annotation.Public;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.java.DataSet;
+import org.apache.flink.util.Preconditions;
 
 /**
  * Base class for operations that operates on two input data sets.
@@ -37,8 +38,8 @@ public abstract class TwoInputOperator<IN1, IN2, OUT, O extends TwoInputOperator
 	
 	
 	protected TwoInputOperator(DataSet<IN1> input1, DataSet<IN2> input2, TypeInformation<OUT> resultType) {
-		super(input1.getExecutionEnvironment(), resultType);
-		
+		super(Preconditions.checkNotNull(input1, "input1 is null").getExecutionEnvironment(), resultType);
+		Preconditions.checkNotNull(input2, "input2 is null");
 		DataSet.checkSameExecutionContext(input1, input2);
 		this.input1 = input1;
 		this.input2 = input2;