[FLINK-3700] [core] Remove Guava dependency from flink-core

Almost all Guava functionality used within flink-core has corresponding
utils in Flink's codebase, or the JDK library.

This replaces the Guava code as follows
  - Preconditions calls by Flink's Preconditions class
  - Collection utils by simple Java Collection calls
  - Iterator's by Flink's Union Iterator
  - Files by simple util methods arount java.nio.Files
  - InetAddresses IPv6 encoding code has been adapted into Flink's NetUtils (with attribution comments)

Some util classes where moved from flink-runtime to flink-core.

This closes #1854

flink-contrib/flink-storm-examples/src/test/java/org/apache/flink/storm/tests/StormFieldsGroupingITCase.java

@@ -21,7 +21,7 @@
 import backtype.storm.topology.TopologyBuilder;
 import backtype.storm.tuple.Fields;
 
-import org.apache.flink.runtime.util.MathUtils;
+import org.apache.flink.util.MathUtils;
 import org.apache.flink.storm.api.FlinkLocalCluster;
 import org.apache.flink.storm.api.FlinkTopology;
 import org.apache.flink.storm.tests.operators.FiniteRandomSpout;

flink-core/pom.xml

@@ -70,12 +70,6 @@ under the License.
 			<artifactId>${shading-artifact.name}</artifactId>
 			<version>${project.version}</version>
 		</dependency>
-
-		<dependency>
-			<groupId>com.google.guava</groupId>
-			<artifactId>guava</artifactId>
-			<version>${guava.version}</version>
-		</dependency>
 
 		<!-- test depedencies -->
 

flink-core/src/main/java/org/apache/flink/api/common/Plan.java

@@ -18,9 +18,6 @@
 
 package org.apache.flink.api.common;
 
-import static com.google.common.base.Preconditions.checkArgument;
-import static com.google.common.base.Preconditions.checkNotNull;
-
 import java.util.ArrayList;
 import java.util.Calendar;
 import java.util.Collection;
@@ -43,6 +40,8 @@
 import org.apache.flink.util.Visitable;
 import org.apache.flink.util.Visitor;
 
+import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkArgument;
 /**
  * This class represents Flink programs, in the form of dataflow plans.
  *

flink-core/src/main/java/org/apache/flink/api/common/TaskInfo.java

@@ -20,8 +20,8 @@
 
 import org.apache.flink.annotation.Internal;
 
-import static com.google.common.base.Preconditions.checkArgument;
-import static com.google.common.base.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkArgument;
 
 /**
  * Encapsulates task-specific information: name, index of subtask, parallelism and attempt number.

flink-core/src/main/java/org/apache/flink/api/common/functions/util/AbstractRuntimeUDFContext.java

@@ -23,7 +23,6 @@
 import java.util.Map;
 import java.util.concurrent.Future;
 
-import com.google.common.base.Preconditions;
 import org.apache.flink.annotation.PublicEvolving;
 import org.apache.flink.api.common.ExecutionConfig;
 import org.apache.flink.api.common.TaskInfo;
@@ -45,6 +44,8 @@
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.core.fs.Path;
 
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
 /**
  * A standalone implementation of the {@link RuntimeContext}, created by runtime UDF operators.
  */
@@ -66,11 +67,11 @@ public AbstractRuntimeUDFContext(TaskInfo taskInfo,
 										ExecutionConfig executionConfig,
 										Map<String, Accumulator<?,?>> accumulators,
 										Map<String, Future<Path>> cpTasks) {
-		this.taskInfo = Preconditions.checkNotNull(taskInfo);
+		this.taskInfo = checkNotNull(taskInfo);
 		this.userCodeClassLoader = userCodeClassLoader;
 		this.executionConfig = executionConfig;
-		this.distributedCache = new DistributedCache(Preconditions.checkNotNull(cpTasks));
-		this.accumulators = Preconditions.checkNotNull(accumulators);
+		this.distributedCache = new DistributedCache(checkNotNull(cpTasks));
+		this.accumulators = checkNotNull(accumulators);
 	}
 
 	@Override

flink-core/src/main/java/org/apache/flink/api/common/io/DelimitedInputFormat.java

@@ -18,12 +18,7 @@
 
 package org.apache.flink.api.common.io;
 
-import java.io.IOException;
-import java.util.ArrayList;
-
 import org.apache.flink.annotation.Public;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import org.apache.flink.api.common.io.statistics.BaseStatistics;
 import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.configuration.Configuration;
@@ -33,7 +28,12 @@
 import org.apache.flink.core.fs.FileSystem;
 import org.apache.flink.core.fs.Path;
 
-import com.google.common.base.Charsets;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.nio.charset.Charset;
+import java.util.ArrayList;
 
 /**
  * Base implementation for input formats that split the input at a delimiter into records.
@@ -53,6 +53,9 @@ public abstract class DelimitedInputFormat<OT> extends FileInputFormat<OT> {
 	 * The log.
 	 */
 	private static final Logger LOG = LoggerFactory.getLogger(DelimitedInputFormat.class);
+
+	/** The default charset  to convert strings to bytes */
+	private static final Charset UTF_8_CHARSET = Charset.forName("UTF-8");
 
 	/**
 	 * The default read buffer size = 1MB.
@@ -185,7 +188,7 @@ public void setDelimiter(char delimiter) {
 	}
 
 	public void setDelimiter(String delimiter) {
-		this.delimiter = delimiter.getBytes(Charsets.UTF_8);
+		this.delimiter = delimiter.getBytes(UTF_8_CHARSET);
 	}
 
 	public int getLineLengthLimit() {

flink-core/src/main/java/org/apache/flink/api/common/io/FileInputFormat.java

@@ -18,22 +18,10 @@
 
 package org.apache.flink.api.common.io;
 
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-
-import com.google.common.base.Preconditions;
 import org.apache.flink.annotation.Public;
 import org.apache.flink.api.common.io.compression.DeflateInflaterInputStreamFactory;
 import org.apache.flink.api.common.io.compression.GzipInflaterInputStreamFactory;
 import org.apache.flink.api.common.io.compression.InflaterInputStreamFactory;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import org.apache.flink.api.common.io.statistics.BaseStatistics;
 import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.configuration.Configuration;
@@ -45,6 +33,20 @@
 import org.apache.flink.core.fs.FileSystem;
 import org.apache.flink.core.fs.Path;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
 /**
  * The base class for {@link RichInputFormat}s that read from files. For specific input types the
  * {@link #nextRecord(Object)} and {@link #reachedEnd()} methods need to be implemented.
@@ -143,7 +145,7 @@ protected static InflaterInputStreamFactory<?> getInflaterInputStreamFactory(Str
 	 * @return the extension of the file name or {@code null} if there is no extension.
 	 */
 	protected static String extractFileExtension(String fileName) {
-		Preconditions.checkNotNull(fileName);
+		checkNotNull(fileName);
 		int lastPeriodIndex = fileName.lastIndexOf('.');
 		if (lastPeriodIndex < 0){
 			return null;

flink-core/src/main/java/org/apache/flink/api/common/io/GenericCsvInputFormat.java

@@ -16,20 +16,16 @@
  * limitations under the License.
  */
 
-
 package org.apache.flink.api.common.io;
 
-import com.google.common.base.Charsets;
-import com.google.common.base.Preconditions;
-import com.google.common.primitives.Ints;
-
 import org.apache.flink.annotation.Internal;
 import org.apache.flink.core.fs.FileInputSplit;
 import org.apache.flink.core.fs.Path;
 import org.apache.flink.types.parser.FieldParser;
 import org.apache.flink.types.parser.StringParser;
 import org.apache.flink.types.parser.StringValueParser;
 import org.apache.flink.util.InstantiationUtil;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -41,13 +37,20 @@
 import java.util.Map;
 import java.util.TreeMap;
 
+import static org.apache.flink.util.Preconditions.checkArgument;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
 @Internal
 public abstract class GenericCsvInputFormat<OT> extends DelimitedInputFormat<OT> {
 
-	private static final Logger LOG = LoggerFactory.getLogger(GenericCsvInputFormat.class);
-
 	private static final long serialVersionUID = 1L;
 
+
+	private static final Logger LOG = LoggerFactory.getLogger(GenericCsvInputFormat.class);
+
+	/** The default charset  to convert strings to bytes */
+	private static final Charset UTF_8_CHARSET = Charset.forName("UTF-8");
+
 	private static final Class<?>[] EMPTY_TYPES = new Class<?>[0];
 
 	private static final boolean[] EMPTY_INCLUDED = new boolean[0];
@@ -127,7 +130,7 @@ public void setCommentPrefix(char commentPrefix) {
 	}
 
 	public void setCommentPrefix(String commentPrefix) {
-		setCommentPrefix(commentPrefix, Charsets.UTF_8);
+		setCommentPrefix(commentPrefix, UTF_8_CHARSET);
 	}
 
 	public void setCommentPrefix(String commentPrefix, String charsetName) throws IllegalCharsetNameException, UnsupportedCharsetException {
@@ -171,7 +174,7 @@ public void setFieldDelimiter(char delimiter) {
 	}
 
 	public void setFieldDelimiter(String delimiter) {
-		this.fieldDelim = delimiter.getBytes(Charsets.UTF_8);
+		this.fieldDelim = delimiter.getBytes(UTF_8_CHARSET);
 	}
 
 	public boolean isLenient() {
@@ -247,9 +250,9 @@ protected void setFieldTypesGeneric(Class<?> ... fieldTypes) {
 	}
 
 	protected void setFieldsGeneric(int[] sourceFieldIndices, Class<?>[] fieldTypes) {
-		Preconditions.checkNotNull(sourceFieldIndices);
-		Preconditions.checkNotNull(fieldTypes);
-		Preconditions.checkArgument(sourceFieldIndices.length == fieldTypes.length,
+		checkNotNull(sourceFieldIndices);
+		checkNotNull(fieldTypes);
+		checkArgument(sourceFieldIndices.length == fieldTypes.length,
 			"Number of field indices and field types must match.");
 
 		for (int i : sourceFieldIndices) {
@@ -258,7 +261,7 @@ protected void setFieldsGeneric(int[] sourceFieldIndices, Class<?>[] fieldTypes)
 			}
 		}
 
-		int largestFieldIndex = Ints.max(sourceFieldIndices);
+		int largestFieldIndex = max(sourceFieldIndices);
 		this.fieldIncluded = new boolean[largestFieldIndex + 1];
 		ArrayList<Class<?>> types = new ArrayList<Class<?>>();
 
@@ -280,8 +283,8 @@ protected void setFieldsGeneric(int[] sourceFieldIndices, Class<?>[] fieldTypes)
 	}
 
 	protected void setFieldsGeneric(boolean[] includedMask, Class<?>[] fieldTypes) {
-		Preconditions.checkNotNull(includedMask);
-		Preconditions.checkNotNull(fieldTypes);
+		checkNotNull(includedMask);
+		checkNotNull(fieldTypes);
 
 		ArrayList<Class<?>> types = new ArrayList<Class<?>>();
 
@@ -530,4 +533,14 @@ protected static void checkForMonotonousOrder(int[] positions, Class<?>[] types)
 			lastPos = positions[i];
 		}
 	}
+
+	private static int max(int[] ints) {
+		checkArgument(ints.length > 0);
+
+		int max = ints[0];
+		for (int i = 1 ; i < ints.length; i++) {
+			max = Math.max(max, ints[i]);
+		}
+		return max;
+	}
 }

flink-core/src/main/java/org/apache/flink/api/common/operators/GenericDataSinkBase.java

@@ -16,7 +16,6 @@
  * limitations under the License.
  */
 
-
 package org.apache.flink.api.common.operators;
 
 import java.util.List;
@@ -39,7 +38,7 @@
 import org.apache.flink.types.Nothing;
 import org.apache.flink.util.Visitor;
 
-import com.google.common.base.Preconditions;
+import static org.apache.flink.util.Preconditions.checkNotNull;
 
 /**
  * Operator for nodes that act as data sinks, storing the data they receive.
@@ -66,7 +65,7 @@ public class GenericDataSinkBase<IN> extends Operator<Nothing> {
 	public GenericDataSinkBase(OutputFormat<IN> f, UnaryOperatorInformation<IN, Nothing> operatorInfo, String name) {
 		super(operatorInfo, name);
 
-		Preconditions.checkNotNull(f, "The OutputFormat may not be null.");
+		checkNotNull(f, "The OutputFormat may not be null.");
 		this.formatWrapper = new UserCodeObjectWrapper<OutputFormat<IN>>(f);
 	}
 
@@ -79,8 +78,7 @@ public GenericDataSinkBase(OutputFormat<IN> f, UnaryOperatorInformation<IN, Noth
 	 */
 	public GenericDataSinkBase(UserCodeWrapper<? extends OutputFormat<IN>> f, UnaryOperatorInformation<IN, Nothing> operatorInfo, String name) {
 		super(operatorInfo, name);
-		Preconditions.checkNotNull(f, "The OutputFormat class may not be null.");
-		this.formatWrapper = f;
+		this.formatWrapper = checkNotNull(f, "The OutputFormat class may not be null.");
 	}
 
 	// --------------------------------------------------------------------------------------------
@@ -100,8 +98,7 @@ public Operator<IN> getInput() {
 	 * @param input The operator to use as the input.
 	 */
 	public void setInput(Operator<IN> input) {
-		Preconditions.checkNotNull(input, "The input may not be null.");
-		this.input = input;
+		this.input = checkNotNull(input, "The input may not be null.");
 	}
 
 	/**
@@ -112,7 +109,7 @@ public void setInput(Operator<IN> input) {
 	 */
 	@Deprecated
 	public void setInputs(Operator<IN>... inputs) {
-		Preconditions.checkNotNull(inputs, "The inputs may not be null.");
+		checkNotNull(inputs, "The inputs may not be null.");
 		this.input = Operator.createUnionCascade(inputs);
 	}
 
@@ -124,7 +121,7 @@ public void setInputs(Operator<IN>... inputs) {
 	 */
 	@Deprecated
 	public void setInputs(List<Operator<IN>> inputs) {
-		Preconditions.checkNotNull(inputs, "The inputs may not be null.");
+		checkNotNull(inputs, "The inputs may not be null.");
 		this.input = Operator.createUnionCascade(inputs);
 	}
 
@@ -136,7 +133,7 @@ public void setInputs(List<Operator<IN>> inputs) {
 	 */
 	@Deprecated
 	public void addInput(Operator<IN>... inputs) {
-		Preconditions.checkNotNull(inputs, "The input may not be null.");
+		checkNotNull(inputs, "The input may not be null.");
 		this.input = Operator.createUnionCascade(this.input, inputs);
 	}
 
@@ -149,7 +146,7 @@ public void addInput(Operator<IN>... inputs) {
 	@SuppressWarnings("unchecked")
 	@Deprecated
 	public void addInputs(List<? extends Operator<IN>> inputs) {
-		Preconditions.checkNotNull(inputs, "The inputs may not be null.");
+		checkNotNull(inputs, "The inputs may not be null.");
 		this.input = createUnionCascade(this.input, (Operator<IN>[]) inputs.toArray(new Operator[inputs.size()]));
 	}
 
@@ -259,7 +256,7 @@ public int compare(IN o1, IN o2) {
 		format.configure(this.parameters);
 
 		if(format instanceof RichOutputFormat){
-			((RichOutputFormat) format).setRuntimeContext(ctx);
+			((RichOutputFormat<?>) format).setRuntimeContext(ctx);
 		}
 		format.open(0, 1);
 		for (IN element : inputData) {