[FLINK-1251] Enums are now handled properly by the collection input format

flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/EnumSerializer.java

@@ -91,6 +91,16 @@ public final class EnumSerializer<T extends Enum<T>> extends TypeSerializer<T> {
 		target.write(source, 4);
 	}
 
+	@Override
+	public boolean equals(Object obj) {
+		if(obj instanceof EnumSerializer) {
+			EnumSerializer other = (EnumSerializer) obj;
+			return other.enumClass == this.enumClass;
+		} else {
+			return false;
+		}
+	}
+
 	// --------------------------------------------------------------------------------------------
 
 	private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {

flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/ReduceITCase.java

@@ -369,6 +369,8 @@ public class ReduceITCase extends JavaProgramTestBase {
 						throw new RuntimeException("Unexpected value for l=" + l);
 					}
 				});
+				ds = ds.union(CollectionDataSets.getPojoWithDateAndEnum(env));
+
 				DataSet<String> res = ds.groupBy("group").reduceGroup(new GroupReduceFunction<CollectionDataSets.PojoWithDateAndEnum, String>() {
 					private static final long serialVersionUID = 1L;
 
@@ -381,7 +383,7 @@ public class ReduceITCase extends JavaProgramTestBase {
 							} else if(val.cat == CollectionDataSets.Category.CAT_B) {
 								Assert.assertEquals("b", val.group);
 							} else {
-								Assert.fail("error");
+								Assert.fail("error. Cat = "+val.cat);
 							}
 							Assert.assertEquals(666, val.date.getTime());
 						}

flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/util/CollectionDataSets.java

@@ -549,11 +549,11 @@ public class CollectionDataSets {
 		data.add(one);
 		
 		PojoWithDateAndEnum two = new PojoWithDateAndEnum();
-		two.group = "a"; two.date = new Date(666); //two.cat = Category.CAT_A;
+		two.group = "a"; two.date = new Date(666); two.cat = Category.CAT_A;
 		data.add(two);
 		
 		PojoWithDateAndEnum three = new PojoWithDateAndEnum();
-		three.group = "b"; three.date = new Date(666); //three.cat = Category.CAT_B;
+		three.group = "b"; three.date = new Date(666); three.cat = Category.CAT_B;
 		data.add(three);
 		
 		return env.fromCollection(data);