[FLINK-4032] Replace all usage of Guava Preconditions

This closes apache#2084

flink-core/src/main/java/org/apache/flink/util/Preconditions.java

@@ -234,6 +234,23 @@ public static void checkElementIndex(int index, int size) {
  }
  }
 
+ /**
+ * Ensures that the given index is valid for an array, list or string of the given size.
+ *
+ * @param index index to check
+ * @param size size of the array, list or string
+ * @param errorMessage The message for the {@code IndexOutOfBoundsException} that is thrown if the check fails. 
+ *
+ * @throws IllegalArgumentException Thrown, if size is negative.
+ * @throws IndexOutOfBoundsException Thrown, if the index negative or greater than or equal to size
+ */
+ public static void checkElementIndex(int index, int size, @Nullable String errorMessage) {
+ checkArgument(size >= 0, "Size was negative.");
+ if (index < 0 || index >= size) {
+ throw new IndexOutOfBoundsException(String.valueOf(errorMessage) + " Index: " + index + ", Size: " + size);
+ }
+ }
+
  // ------------------------------------------------------------------------
  // Utilities
  // ------------------------------------------------------------------------

flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/aggregate/AggregateMapFunction.scala

@@ -17,12 +17,12 @@
  */
 package org.apache.flink.api.table.runtime.aggregate
 
-import com.google.common.base.Preconditions
 import org.apache.flink.api.common.functions.RichMapFunction
 import org.apache.flink.api.common.typeinfo.TypeInformation
 import org.apache.flink.api.java.typeutils.ResultTypeQueryable
 import org.apache.flink.api.table.Row
 import org.apache.flink.configuration.Configuration
+import org.apache.flink.util.Preconditions
 
 class AggregateMapFunction[IN, OUT](
  private val aggregates: Array[Aggregate[_]],

flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/aggregate/AggregateReduceGroupFunction.scala

@@ -19,11 +19,11 @@ package org.apache.flink.api.table.runtime.aggregate
 
 import java.lang.Iterable
 
-import com.google.common.base.Preconditions
 import org.apache.flink.api.common.functions.{CombineFunction, RichGroupReduceFunction, RichMapPartitionFunction}
 import org.apache.flink.api.table.Row
 import org.apache.flink.configuration.Configuration
 import org.apache.flink.util.Collector
+import org.apache.flink.util.Preconditions
 
 import scala.collection.JavaConversions._
 

flink-optimizer/src/main/java/org/apache/flink/optimizer/plan/Channel.java

@@ -36,7 +36,7 @@
 import org.apache.flink.runtime.operators.shipping.ShipStrategyType;
 import org.apache.flink.runtime.operators.util.LocalStrategy;
 
-import static com.google.common.base.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkNotNull;
 
 /**
  * A Channel represents the result produced by an operator and the data exchange

flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/BackPressureStatsTracker.java

@@ -41,8 +41,8 @@
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
 
-import static com.google.common.base.Preconditions.checkArgument;
-import static com.google.common.base.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkArgument;
+import static org.apache.flink.util.Preconditions.checkNotNull;
 
 /**
  * Back pressure statistics tracker.

flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/ExecutionGraphHolder.java

@@ -31,7 +31,7 @@
 
 import java.util.WeakHashMap;
 
-import static com.google.common.base.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkNotNull;
 
 /**
  * Gateway to obtaining an {@link ExecutionGraph} from a source, like JobManager or Archive.
@@ -95,4 +95,4 @@ else if (result instanceof JobManagerMessages.JobFound) {
  throw new RuntimeException("Error requesting execution graph", e);
  }
  }
-}
+}

flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/JobManagerRetriever.java

@@ -42,7 +42,7 @@
 import java.util.UUID;
 import java.util.concurrent.TimeoutException;
 
-import static com.google.common.base.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkNotNull;
 
 /**
  * Retrieves and stores the actor gateway to the current leading JobManager. In case of an error,

flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/OperatorBackPressureStats.java

@@ -20,8 +20,8 @@
 
 import java.util.Arrays;
 
-import static com.google.common.base.Preconditions.checkArgument;
-import static com.google.common.base.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkArgument;
+import static org.apache.flink.util.Preconditions.checkNotNull;
 
 /**
  * Back pressure statistics of multiple tasks.

flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/RuntimeMonitorHandler.java

@@ -43,7 +43,7 @@
 import java.util.HashMap;
 import java.util.Map;
 
-import static com.google.common.base.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkNotNull;
 
 /**
  * The Netty channel handler that processes all HTTP requests.

flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/RuntimeMonitorHandlerBase.java

@@ -33,7 +33,7 @@
 import scala.concurrent.Future;
 import scala.concurrent.duration.FiniteDuration;
 
-import static com.google.common.base.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkNotNull;
 
 /**
  * The Netty channel handler that processes all HTTP requests.

flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/StackTraceSample.java

@@ -24,7 +24,7 @@
 import java.util.List;
 import java.util.Map;
 
-import static com.google.common.base.Preconditions.checkArgument;
+import static org.apache.flink.util.Preconditions.checkArgument;
 
 /**
  * A sample of stack traces for one or more tasks.

flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/StackTraceSampleCoordinator.java

@@ -49,8 +49,8 @@
 import java.util.TimerTask;
 import java.util.UUID;
 
-import static com.google.common.base.Preconditions.checkArgument;
-import static com.google.common.base.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkArgument;
+import static org.apache.flink.util.Preconditions.checkNotNull;
 
 /**
  * A coordinator for triggering and collecting stack traces of running tasks.

flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/WebRuntimeMonitor.java

@@ -83,7 +83,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
-import static com.google.common.base.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkNotNull;
 
 /**
  * The root component of the web runtime monitor. This class starts the web server and creates

flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/files/StaticFileServerHandler.java

@@ -72,7 +72,6 @@
 import java.util.Locale;
 import java.util.TimeZone;
 
-import static com.google.common.base.Preconditions.checkNotNull;
 import static io.netty.handler.codec.http.HttpHeaders.Names.CACHE_CONTROL;
 import static io.netty.handler.codec.http.HttpHeaders.Names.CONNECTION;
 import static io.netty.handler.codec.http.HttpHeaders.Names.CONTENT_TYPE;
@@ -85,6 +84,7 @@
 import static io.netty.handler.codec.http.HttpResponseStatus.NOT_MODIFIED;
 import static io.netty.handler.codec.http.HttpResponseStatus.OK;
 import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1;
+import static org.apache.flink.util.Preconditions.checkNotNull;
 
 /**
  * Simple file server handler that serves requests to web frontend's static files, such as

flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/ClusterOverviewHandler.java

@@ -30,7 +30,7 @@
 import java.io.StringWriter;
 import java.util.Map;
 
-import static com.google.common.base.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkNotNull;
 
 /**
  * Responder that returns the status of the Flink cluster, such as how many

flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/CurrentJobsOverviewHandler.java

@@ -31,7 +31,7 @@
 import java.io.StringWriter;
 import java.util.Map;
 
-import static com.google.common.base.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkNotNull;
 
 /**
  * Request handler that returns a summary of the job status.

flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/HandlerRedirectUtils.java

@@ -37,7 +37,7 @@
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
-import static com.google.common.base.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkNotNull;
 
 /**
  * Utilities to extract a redirect address.

flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobVertexBackPressureHandler.java

@@ -28,8 +28,8 @@
 import java.io.StringWriter;
 import java.util.Map;
 
-import static com.google.common.base.Preconditions.checkArgument;
-import static com.google.common.base.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkArgument;
+import static org.apache.flink.util.Preconditions.checkNotNull;
 
 /**
  * Request handler that returns back pressure stats for a single job vertex and

flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/TaskManagerLogHandler.java

@@ -71,11 +71,11 @@
 import java.util.HashMap;
 import java.util.concurrent.ConcurrentHashMap;
 
-import static com.google.common.base.Preconditions.checkNotNull;
 import static io.netty.handler.codec.http.HttpHeaders.Names.CONNECTION;
 import static io.netty.handler.codec.http.HttpHeaders.Names.CONTENT_TYPE;
 import static io.netty.handler.codec.http.HttpResponseStatus.OK;
 import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1;
+import static org.apache.flink.util.Preconditions.checkNotNull;
 
 /**
  * Request handler that returns the TaskManager log/out files.

flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobServer.java

@@ -42,7 +42,7 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 
-import static com.google.common.base.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkNotNull;
 
 /**
  * This class implements the BLOB server. The BLOB server is responsible for listening for incoming requests and

flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobUtils.java

@@ -39,7 +39,7 @@
 import java.security.NoSuchAlgorithmException;
 import java.util.UUID;
 
-import static com.google.common.base.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkNotNull;
 
 /**
  * Utility class to work with blob data.

flink-runtime/src/main/java/org/apache/flink/runtime/blob/FileSystemBlobStore.java

@@ -36,7 +36,7 @@
 import java.io.OutputStream;
 import java.net.URI;
 
-import static com.google.common.base.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkNotNull;
 
 /**
  * Blob store backed by {@link FileSystem}.

flink-runtime/src/main/java/org/apache/flink/runtime/broadcast/BroadcastVariableMaterialization.java

@@ -31,11 +31,10 @@
 import org.apache.flink.runtime.operators.BatchTask;
 import org.apache.flink.runtime.operators.util.ReaderIterator;
 import org.apache.flink.runtime.plugable.DeserializationDelegate;
+import org.apache.flink.util.Preconditions;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import com.google.common.base.Preconditions;
-
 /**
  * @param <T> The type of the elements in the broadcasted data set.
  */

flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java

@@ -55,8 +55,8 @@
 import java.util.TimerTask;
 import java.util.UUID;
 
-import static com.google.common.base.Preconditions.checkArgument;
-import static com.google.common.base.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkArgument;
+import static org.apache.flink.util.Preconditions.checkNotNull;
 
 /**
  * The checkpoint coordinator coordinates the distributed snapshots of operators and state.

flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorDeActivator.java

@@ -18,10 +18,10 @@
 
 package org.apache.flink.runtime.checkpoint;
 
-import com.google.common.base.Preconditions;
 import org.apache.flink.runtime.akka.FlinkUntypedActor;
 import org.apache.flink.runtime.jobgraph.JobStatus;
 import org.apache.flink.runtime.messages.ExecutionGraphMessages;
+import org.apache.flink.util.Preconditions;
 
 import java.util.UUID;
 

flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/FileSystemStateStore.java

@@ -28,7 +28,7 @@
 import java.io.ObjectOutputStream;
 import java.io.Serializable;
 
-import static com.google.common.base.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkNotNull;
 
 /**
  * {@link FileSystem} backed {@link StateStore}.

flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/HeapStateStore.java

@@ -24,7 +24,7 @@
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.atomic.AtomicInteger;
 
-import static com.google.common.base.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkNotNull;
 
 /**
  * Java heap backed {@link StateStore}.

flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SavepointCoordinator.java

@@ -44,7 +44,7 @@
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
 
-import static com.google.common.base.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkNotNull;
 
 /**
  * The savepoint coordinator is a slightly modified variant of the regular

flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SavepointCoordinatorDeActivator.java

@@ -18,10 +18,10 @@
 
 package org.apache.flink.runtime.checkpoint;
 
-import com.google.common.base.Preconditions;
 import org.apache.flink.runtime.akka.FlinkUntypedActor;
 import org.apache.flink.runtime.jobgraph.JobStatus;
 import org.apache.flink.runtime.messages.ExecutionGraphMessages;
+import org.apache.flink.util.Preconditions;
 
 import java.util.UUID;
 

flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/StandaloneCompletedCheckpointStore.java

@@ -24,8 +24,8 @@
 import java.util.ArrayList;
 import java.util.List;
 
-import static com.google.common.base.Preconditions.checkArgument;
-import static com.google.common.base.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkArgument;
+import static org.apache.flink.util.Preconditions.checkNotNull;
 
 /**
  * {@link CompletedCheckpointStore} for JobManagers running in {@link RecoveryMode#STANDALONE}.

flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SubtaskState.java

@@ -25,7 +25,7 @@
 
 import java.io.Serializable;
 
-import static com.google.common.base.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkNotNull;
 
 /**
  * Simple bean to describe the state belonging to a parallel operator. It is part of the

flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/ZooKeeperCheckpointIDCounter.java

@@ -27,7 +27,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import static com.google.common.base.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkNotNull;
 
 /**
  * {@link CheckpointIDCounter} instances for JobManagers running in {@link RecoveryMode#ZOOKEEPER}.

flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/ZooKeeperCheckpointRecoveryFactory.java

@@ -24,7 +24,7 @@
 import org.apache.flink.runtime.jobmanager.RecoveryMode;
 import org.apache.flink.runtime.util.ZooKeeperUtils;
 
-import static com.google.common.base.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkNotNull;
 
 /**
  * {@link CheckpointCoordinator} components in {@link RecoveryMode#ZOOKEEPER}.

flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStore.java

@@ -36,8 +36,8 @@
 import java.util.ConcurrentModificationException;
 import java.util.List;
 
-import static com.google.common.base.Preconditions.checkArgument;
-import static com.google.common.base.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkArgument;
+import static org.apache.flink.util.Preconditions.checkNotNull;
 
 /**
  * {@link CompletedCheckpointStore} for JobManagers running in {@link RecoveryMode#ZOOKEEPER}.

flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/stats/OperatorCheckpointStats.java

@@ -20,7 +20,7 @@
 
 import java.util.Arrays;
 
-import static com.google.common.base.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkNotNull;
 
 /**
  * Statistics for a specific checkpoint per operator.

flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/stats/SimpleCheckpointStatsTracker.java

@@ -31,7 +31,7 @@
 import java.util.List;
 import java.util.Map;
 
-import static com.google.common.base.Preconditions.checkArgument;
+import static org.apache.flink.util.Preconditions.checkArgument;
 
 /**
  * A simple checkpoint stats tracker.

flink-runtime/src/main/java/org/apache/flink/runtime/client/JobClient.java

@@ -54,7 +54,7 @@
 import java.net.InetSocketAddress;
 import java.util.concurrent.TimeoutException;
 
-import static com.google.common.base.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkNotNull;
 
 /**
  * The JobClient bridges between the JobManager's asynchronous actor messages and