[FLINK-6943] Improve exceptions within TypeExtractionUtils#getSingleA…

…bstractMethod

This closes apache#4140.

flink-core/src/main/java/org/apache/flink/api/java/typeutils/TypeExtractionUtils.java

@@ -215,22 +215,33 @@ public static Type extractTypeArgument(Type t, int index) throws InvalidTypesExc
 	 * Extracts a Single Abstract Method (SAM) as defined in Java Specification (4.3.2. The Class Object,
 	 * 9.8 Functional Interfaces, 9.4.3 Interface Method Body) from given class.
 	 *
-	 * @param baseClass
-	 * @throws InvalidTypesException if the given class does not implement
-	 * @return
+	 * @param baseClass a class that is a FunctionalInterface to retrieve a SAM from
+	 * @throws InvalidTypesException if the given class does not implement FunctionalInterface
+	 * @return single abstract method of the given class
 	 */
 	public static Method getSingleAbstractMethod(Class<?> baseClass) {
+
+		if (!baseClass.isInterface()) {
+			throw new InvalidTypesException("Given class: " + baseClass + "is not a FunctionalInterface.");
+		}
+
 		Method sam = null;
 		for (Method method : baseClass.getMethods()) {
 			if (Modifier.isAbstract(method.getModifiers())) {
 				if (sam == null) {
 					sam = method;
 				} else {
-					throw new InvalidTypesException(
-						"Given class: " + baseClass + " is not a FunctionalInterface. It does not have a SAM.");
+					throw new InvalidTypesException("Given class: " + baseClass +
+						" is not a FunctionalInterface. It has more than one abstract method.");
 				}
 			}
 		}
+
+		if (sam == null) {
+			throw new InvalidTypesException(
+				"Given class: " + baseClass + " is not a FunctionalInterface. It does not have any abstract methods.");
+		}
+
 		return sam;
 	}
 

flink-java8/src/test/java/org/apache/flink/api/java/type/lambdas/LambdaExtractionTest.java

@@ -23,6 +23,7 @@
 import org.apache.flink.api.common.functions.FlatJoinFunction;
 import org.apache.flink.api.common.functions.FlatMapFunction;
 import org.apache.flink.api.common.functions.GroupReduceFunction;
+import org.apache.flink.api.common.functions.InvalidTypesException;
 import org.apache.flink.api.common.functions.JoinFunction;
 import org.apache.flink.api.common.functions.MapFunction;
 import org.apache.flink.api.common.functions.MapPartitionFunction;
@@ -35,13 +36,17 @@
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.api.java.typeutils.MissingTypeInfo;
 import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.api.java.typeutils.TypeExtractionUtils;
 import org.apache.flink.api.java.typeutils.TypeExtractor;
 import org.apache.flink.api.java.typeutils.TypeInfoParser;
 
 import org.junit.Assert;
 import org.junit.Test;
 
+import java.lang.reflect.Method;
+
 import static org.apache.flink.api.java.typeutils.TypeExtractionUtils.checkAndExtractLambda;
+import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.fail;
@@ -322,4 +327,50 @@ public void testConstructorMethodRef() {
 		Assert.assertEquals(BasicTypeInfo.INT_TYPE_INFO, ti);
 	}
 
+	public interface InterfaceWithDefaultMethod {
+		void samMethod();
+
+		default void defaultMethod() {
+
+		}
+	}
+
+	@Test
+	public void testSamMethodExtractionInterfaceWithDefaultMethod() {
+		final Method sam = TypeExtractionUtils.getSingleAbstractMethod(InterfaceWithDefaultMethod.class);
+		assertNotNull(sam);
+		assertEquals("samMethod", sam.getName());
+	}
+
+	public interface InterfaceWithMultipleMethods {
+		void firstMethod();
+
+		void secondMethod();
+	}
+
+	@Test(expected = InvalidTypesException.class)
+	public void getSingleAbstractMethodMultipleMethods() throws Exception {
+		TypeExtractionUtils.getSingleAbstractMethod(InterfaceWithMultipleMethods.class);
+	}
+
+	public interface InterfaceWithoutAbstractMethod {
+		default void defaultMethod() {
+
+		};
+	}
+
+	@Test(expected = InvalidTypesException.class)
+	public void getSingleAbstractMethodNoAbstractMethods() throws Exception {
+		TypeExtractionUtils.getSingleAbstractMethod(InterfaceWithoutAbstractMethod.class);
+	}
+
+	public abstract class AbstractClassWithSingleAbstractMethod {
+		public abstract void defaultMethod();
+	}
+
+	@Test(expected = InvalidTypesException.class)
+	public void getSingleAbstractMethodNotAnInterface() throws Exception {
+		TypeExtractionUtils.getSingleAbstractMethod(AbstractClassWithSingleAbstractMethod.class);
+	}
+
 }