[FLINK-7190] [java] Activate checkstyle flink-java/*

This closes apache#4343.

Author: dawidwys

flink-java/src/main/java/org/apache/flink/api/java/ClosureCleaner.java

@@ -26,7 +26,6 @@
 import org.objectweb.asm.ClassVisitor;
 import org.objectweb.asm.MethodVisitor;
 import org.objectweb.asm.Opcodes;
-
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -41,53 +40,52 @@
  */
 @Internal
 public class ClosureCleaner {
-	
-	private static Logger LOG = LoggerFactory.getLogger(ClosureCleaner.class);
-	
+
+	private static final Logger LOG = LoggerFactory.getLogger(ClosureCleaner.class);
+
 	/**
 	 * Tries to clean the closure of the given object, if the object is a non-static inner
 	 * class.
-	 * 
+	 *
 	 * @param func The object whose closure should be cleaned.
 	 * @param checkSerializable Flag to indicate whether serializability should be checked after
 	 *                          the closure cleaning attempt.
-	 * 
+	 *
 	 * @throws InvalidProgramException Thrown, if 'checkSerializable' is true, and the object was
 	 *                                 not serializable after the closure cleaning.
-	 * 
+	 *
 	 * @throws RuntimeException A RuntimeException may be thrown, if the code of the class could not
 	 *                          be loaded, in order to process during teh closure cleaning.
 	 */
 	public static void clean(Object func, boolean checkSerializable) {
 		if (func == null) {
 			return;
 		}
-		
+
 		final Class<?> cls = func.getClass();
 
 		// First find the field name of the "this$0" field, this can
 		// be "this$x" depending on the nesting
 		boolean closureAccessed = false;
-		
+
 		for (Field f: cls.getDeclaredFields()) {
 			if (f.getName().startsWith("this$")) {
 				// found a closure referencing field - now try to clean
 				closureAccessed |= cleanThis0(func, cls, f.getName());
 			}
 		}
-		
+
 		if (checkSerializable) {
 			try {
 				InstantiationUtil.serializeObject(func);
 			}
 			catch (Exception e) {
 				String functionType = getSuperClassOrInterfaceName(func.getClass());
-				
+
 				String msg = functionType == null ?
 						(func + " is not serializable.") :
 						("The implementation of the " + functionType + " is not serializable.");
-				
-				
+
 				if (closureAccessed) {
 					msg += " The implementation accesses fields of its enclosing class, which is " +
 							"a common reason for non-serializability. " +
@@ -96,7 +94,7 @@ public static void clean(Object func, boolean checkSerializable) {
 				} else {
 					msg += " The object probably contains or references non serializable fields.";
 				}
-				
+
 				throw new InvalidProgramException(msg, e);
 			}
 		}
@@ -109,14 +107,14 @@ public static void ensureSerializable(Object obj) {
 			throw new InvalidProgramException("Object " + obj + " is not serializable", e);
 		}
 	}
-	
+
 	private static boolean cleanThis0(Object func, Class<?> cls, String this0Name) {
-		
+
 		This0AccessFinder this0Finder = new This0AccessFinder(this0Name);
 		getClassReader(cls).accept(this0Finder, 0);
-		
+
 		final boolean accessesClosure = this0Finder.isThis0Accessed();
-				
+
 		if (LOG.isDebugEnabled()) {
 			LOG.debug(this0Name + " is accessed: " + accessesClosure);
 		}
@@ -129,7 +127,7 @@ private static boolean cleanThis0(Object func, Class<?> cls, String this0Name) {
 				// has no this$0, just return
 				throw new RuntimeException("Could not set " + this0Name + ": " + e);
 			}
-			
+
 			try {
 				this0.setAccessible(true);
 				this0.set(func, null);
@@ -139,10 +137,10 @@ private static boolean cleanThis0(Object func, Class<?> cls, String this0Name) {
 				throw new RuntimeException("Could not set " + this0Name + " to null. " + e.getMessage(), e);
 			}
 		}
-		
+
 		return accessesClosure;
 	}
-	
+
 	private static ClassReader getClassReader(Class<?> cls) {
 		String className = cls.getName().replaceFirst("^.*\\.", "") + ".class";
 		try {
@@ -151,8 +149,7 @@ private static ClassReader getClassReader(Class<?> cls) {
 			throw new RuntimeException("Could not create ClassReader: " + e.getMessage(), e);
 		}
 	}
-	
-	
+
 	private static String getSuperClassOrInterfaceName(Class<?> cls) {
 		Class<?> superclass = cls.getSuperclass();
 		if (superclass.getName().startsWith("org.apache.flink")) {
@@ -176,7 +173,6 @@ class This0AccessFinder extends ClassVisitor {
 
 	private final String this0Name;
 	private boolean isThis0Accessed;
-	
 
 	public This0AccessFinder(String this0Name) {
 		super(Opcodes.ASM5);

flink-java/src/main/java/org/apache/flink/api/java/CollectionEnvironment.java

@@ -23,6 +23,9 @@
 import org.apache.flink.api.common.Plan;
 import org.apache.flink.api.common.operators.CollectionExecutor;
 
+/**
+ * Version of {@link ExecutionEnvironment} that allows serial, local, collection-based executions of Flink programs.
+ */
 @PublicEvolving
 public class CollectionEnvironment extends ExecutionEnvironment {
 
@@ -40,7 +43,7 @@ public JobExecutionResult execute(String jobName) throws Exception {
 	public int getParallelism() {
 		return 1; // always serial
 	}
-	
+
 	@Override
 	public String getExecutionPlan() throws Exception {
 		throw new UnsupportedOperationException("Execution plans are not used for collection-based execution.");