[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 Project: http://git-wip-us.apache.org/repos/asf/flink/repo Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/760a0d9e Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/760a0d9e Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/760a0d9e Branch: refs/heads/master Commit: 760a0d9e7fd9fa88e9f7408b137d78df384d764f Parents: 1b93b32 Author: Stephan Ewen <se...@apache.org> Authored: Tue Apr 5 15:18:32 2016 +0200 Committer: Fabian Hueske <fhue...@apache.org> Committed: Fri Apr 15 19:38:14 2016 +0200 ---------------------------------------------------------------------- .../storm/tests/StormFieldsGroupingITCase.java | 2 +- flink-core/pom.xml | 6 - .../java/org/apache/flink/api/common/Plan.java | 5 +- .../org/apache/flink/api/common/TaskInfo.java | 4 +- .../util/AbstractRuntimeUDFContext.java | 9 +- .../api/common/io/DelimitedInputFormat.java | 17 +- .../flink/api/common/io/FileInputFormat.java | 28 +-- .../api/common/io/GenericCsvInputFormat.java | 43 ++-- .../common/operators/GenericDataSinkBase.java | 21 +- .../apache/flink/api/common/operators/Keys.java | 19 +- .../base/GroupCombineOperatorBase.java | 6 +- .../operators/base/GroupReduceOperatorBase.java | 6 +- .../api/common/operators/util/FieldList.java | 6 +- .../operators/util/UserCodeObjectWrapper.java | 7 +- .../api/common/typeinfo/BasicArrayTypeInfo.java | 10 +- .../api/common/typeinfo/BasicTypeInfo.java | 8 +- .../api/common/typeinfo/FractionalTypeInfo.java | 19 +- .../api/common/typeinfo/IntegerTypeInfo.java | 25 ++- .../api/common/typeinfo/NumericTypeInfo.java | 29 +-- .../common/typeinfo/PrimitiveArrayTypeInfo.java | 22 +- .../api/common/typeutils/CompositeType.java | 4 +- .../common/typeutils/base/EnumSerializer.java | 5 +- .../typeutils/base/GenericArraySerializer.java | 7 +- .../flink/api/java/typeutils/EnumTypeInfo.java | 7 +- .../api/java/typeutils/GenericTypeInfo.java | 4 +- .../api/java/typeutils/ObjectArrayTypeInfo.java | 15 +- .../flink/api/java/typeutils/PojoField.java | 7 +- .../flink/api/java/typeutils/PojoTypeInfo.java | 40 ++-- .../flink/api/java/typeutils/TupleTypeInfo.java | 32 ++- .../api/java/typeutils/TupleTypeInfoBase.java | 6 +- .../flink/api/java/typeutils/TypeExtractor.java | 8 +- .../flink/api/java/typeutils/ValueTypeInfo.java | 12 +- .../api/java/typeutils/WritableTypeInfo.java | 11 +- .../java/typeutils/runtime/AvroSerializer.java | 7 +- .../runtime/CopyableValueSerializer.java | 4 +- .../java/typeutils/runtime/PojoSerializer.java | 10 +- .../typeutils/runtime/TupleSerializerBase.java | 6 +- .../java/typeutils/runtime/ValueSerializer.java | 5 +- .../typeutils/runtime/kryo/KryoSerializer.java | 8 +- .../org/apache/flink/types/StringValue.java | 14 +- .../java/org/apache/flink/util/FileUtils.java | 91 ++++++++ .../java/org/apache/flink/util/IOUtils.java | 223 +++++++++++++++++++ .../java/org/apache/flink/util/MathUtils.java | 175 +++++++++++++++ .../java/org/apache/flink/util/NetUtils.java | 86 ++++++- .../org/apache/flink/util/Preconditions.java | 57 ++++- .../org/apache/flink/util/UnionIterator.java | 102 +++++++++ .../org/apache/flink/util/XORShiftRandom.java | 4 +- .../base/OuterJoinOperatorBaseTest.java | 4 +- .../flink/api/java/tuple/TupleGenerator.java | 7 +- .../java/typeutils/PojoTypeExtractionTest.java | 13 +- .../typeutils/runtime/PojoSerializerTest.java | 8 +- .../runtime/PojoSubclassSerializerTest.java | 5 +- .../SubclassFromInterfaceSerializerTest.java | 5 +- .../apache/flink/testutils/junit/RetryRule.java | 5 +- .../org/apache/flink/util/MathUtilTest.java | 102 +++++++++ .../org/apache/flink/util/NetUtilsTest.java | 4 - .../apache/flink/util/UnionIteratorTest.java | 143 ++++++++++++ .../apache/flink/runtime/blob/BlobUtils.java | 2 +- .../flink/runtime/blob/FileSystemBlobStore.java | 2 +- .../checkpoint/FileSystemStateStore.java | 2 +- .../flink/runtime/filecache/FileCache.java | 2 +- .../runtime/io/disk/FileChannelInputView.java | 2 +- .../runtime/io/disk/RandomAccessInputView.java | 2 +- .../runtime/io/disk/RandomAccessOutputView.java | 2 +- .../io/disk/SeekableFileChannelInputView.java | 2 +- .../io/disk/SimpleCollectingOutputView.java | 2 +- .../flink/runtime/memory/MemoryManager.java | 2 +- .../operators/hash/CompactingHashTable.java | 2 +- .../runtime/operators/hash/HashPartition.java | 2 +- .../operators/hash/MutableHashTable.java | 2 +- .../operators/shipping/OutputEmitter.java | 2 +- .../apache/flink/runtime/util/FileUtils.java | 61 ----- .../org/apache/flink/runtime/util/IOUtils.java | 213 ------------------ .../apache/flink/runtime/util/MathUtils.java | 176 --------------- .../flink/runtime/util/UnionIterator.java | 100 --------- .../FileSystemStateStorageHelper.java | 2 +- .../flink/runtime/taskmanager/TaskManager.scala | 2 +- .../runtime/testutils/CommonTestUtils.java | 2 +- .../apache/flink/runtime/util/MathUtilTest.java | 101 --------- .../flink/runtime/util/UnionIteratorTest.java | 142 ------------ .../source/SocketTextStreamFunction.java | 2 +- ...ractAlignedProcessingTimeWindowOperator.java | 2 +- .../windowing/AccumulatingKeyedTimePanes.java | 2 +- .../runtime/operators/windowing/KeyMap.java | 2 +- .../runtime/partitioner/HashPartitioner.java | 2 +- .../apache/flink/streaming/api/IterateTest.java | 2 +- .../streaming/api/StreamingOperatorsITCase.java | 2 +- .../api/scala/StreamingOperatorsITCase.scala | 2 +- 88 files changed, 1294 insertions(+), 1087 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/flink/blob/760a0d9e/flink-contrib/flink-storm-examples/src/test/java/org/apache/flink/storm/tests/StormFieldsGroupingITCase.java ---------------------------------------------------------------------- diff --git a/flink-contrib/flink-storm-examples/src/test/java/org/apache/flink/storm/tests/StormFieldsGroupingITCase.java b/flink-contrib/flink-storm-examples/src/test/java/org/apache/flink/storm/tests/StormFieldsGroupingITCase.java index 1875ecb..b43b24d 100644 --- a/flink-contrib/flink-storm-examples/src/test/java/org/apache/flink/storm/tests/StormFieldsGroupingITCase.java +++ b/flink-contrib/flink-storm-examples/src/test/java/org/apache/flink/storm/tests/StormFieldsGroupingITCase.java @@ -21,7 +21,7 @@ import backtype.storm.Config; 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; http://git-wip-us.apache.org/repos/asf/flink/blob/760a0d9e/flink-core/pom.xml ---------------------------------------------------------------------- diff --git a/flink-core/pom.xml b/flink-core/pom.xml index 59b620b..eb55bdd 100644 --- a/flink-core/pom.xml +++ b/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 --> http://git-wip-us.apache.org/repos/asf/flink/blob/760a0d9e/flink-core/src/main/java/org/apache/flink/api/common/Plan.java ---------------------------------------------------------------------- diff --git a/flink-core/src/main/java/org/apache/flink/api/common/Plan.java b/flink-core/src/main/java/org/apache/flink/api/common/Plan.java index 30f2c2f..0235af0 100644 --- a/flink-core/src/main/java/org/apache/flink/api/common/Plan.java +++ b/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.core.fs.Path; 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. * http://git-wip-us.apache.org/repos/asf/flink/blob/760a0d9e/flink-core/src/main/java/org/apache/flink/api/common/TaskInfo.java ---------------------------------------------------------------------- diff --git a/flink-core/src/main/java/org/apache/flink/api/common/TaskInfo.java b/flink-core/src/main/java/org/apache/flink/api/common/TaskInfo.java index 6482cde..ac87e74 100644 --- a/flink-core/src/main/java/org/apache/flink/api/common/TaskInfo.java +++ b/flink-core/src/main/java/org/apache/flink/api/common/TaskInfo.java @@ -20,8 +20,8 @@ package org.apache.flink.api.common; 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. http://git-wip-us.apache.org/repos/asf/flink/blob/760a0d9e/flink-core/src/main/java/org/apache/flink/api/common/functions/util/AbstractRuntimeUDFContext.java ---------------------------------------------------------------------- diff --git a/flink-core/src/main/java/org/apache/flink/api/common/functions/util/AbstractRuntimeUDFContext.java b/flink-core/src/main/java/org/apache/flink/api/common/functions/util/AbstractRuntimeUDFContext.java index afe9f77..74b78df 100644 --- a/flink-core/src/main/java/org/apache/flink/api/common/functions/util/AbstractRuntimeUDFContext.java +++ b/flink-core/src/main/java/org/apache/flink/api/common/functions/util/AbstractRuntimeUDFContext.java @@ -23,7 +23,6 @@ import java.util.Collections; 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.state.ValueStateDescriptor; 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 abstract class AbstractRuntimeUDFContext implements RuntimeContext { 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 http://git-wip-us.apache.org/repos/asf/flink/blob/760a0d9e/flink-core/src/main/java/org/apache/flink/api/common/io/DelimitedInputFormat.java ---------------------------------------------------------------------- diff --git a/flink-core/src/main/java/org/apache/flink/api/common/io/DelimitedInputFormat.java b/flink-core/src/main/java/org/apache/flink/api/common/io/DelimitedInputFormat.java index f6b6d49..243e2a4 100644 --- a/flink-core/src/main/java/org/apache/flink/api/common/io/DelimitedInputFormat.java +++ b/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.FileStatus; 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 abstract class DelimitedInputFormat<OT> extends FileInputFormat<OT> { } public void setDelimiter(String delimiter) { - this.delimiter = delimiter.getBytes(Charsets.UTF_8); + this.delimiter = delimiter.getBytes(UTF_8_CHARSET); } public int getLineLengthLimit() { http://git-wip-us.apache.org/repos/asf/flink/blob/760a0d9e/flink-core/src/main/java/org/apache/flink/api/common/io/FileInputFormat.java ---------------------------------------------------------------------- diff --git a/flink-core/src/main/java/org/apache/flink/api/common/io/FileInputFormat.java b/flink-core/src/main/java/org/apache/flink/api/common/io/FileInputFormat.java index c4cd2b3..fd69cc3 100644 --- a/flink-core/src/main/java/org/apache/flink/api/common/io/FileInputFormat.java +++ b/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.FileStatus; 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 @@ public abstract class FileInputFormat<OT> extends RichInputFormat<OT, FileInputS * @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; http://git-wip-us.apache.org/repos/asf/flink/blob/760a0d9e/flink-core/src/main/java/org/apache/flink/api/common/io/GenericCsvInputFormat.java ---------------------------------------------------------------------- diff --git a/flink-core/src/main/java/org/apache/flink/api/common/io/GenericCsvInputFormat.java b/flink-core/src/main/java/org/apache/flink/api/common/io/GenericCsvInputFormat.java index cb20f81..31d42ff 100644 --- a/flink-core/src/main/java/org/apache/flink/api/common/io/GenericCsvInputFormat.java +++ b/flink-core/src/main/java/org/apache/flink/api/common/io/GenericCsvInputFormat.java @@ -16,13 +16,8 @@ * 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; @@ -30,6 +25,7 @@ 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.ArrayList; 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 abstract class GenericCsvInputFormat<OT> extends DelimitedInputFormat<OT> } 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 abstract class GenericCsvInputFormat<OT> extends DelimitedInputFormat<OT> } 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 @@ public abstract class GenericCsvInputFormat<OT> extends DelimitedInputFormat<OT> } 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 @@ public abstract class GenericCsvInputFormat<OT> extends DelimitedInputFormat<OT> } } - 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 @@ public abstract class GenericCsvInputFormat<OT> extends DelimitedInputFormat<OT> } 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 @@ public abstract class GenericCsvInputFormat<OT> extends DelimitedInputFormat<OT> 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; + } } http://git-wip-us.apache.org/repos/asf/flink/blob/760a0d9e/flink-core/src/main/java/org/apache/flink/api/common/operators/GenericDataSinkBase.java ---------------------------------------------------------------------- diff --git a/flink-core/src/main/java/org/apache/flink/api/common/operators/GenericDataSinkBase.java b/flink-core/src/main/java/org/apache/flink/api/common/operators/GenericDataSinkBase.java index 33f11f3..1f3875d 100644 --- a/flink-core/src/main/java/org/apache/flink/api/common/operators/GenericDataSinkBase.java +++ b/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.api.common.typeutils.TypeComparator; 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 class GenericDataSinkBase<IN> extends Operator<Nothing> { */ 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 class GenericDataSinkBase<IN> extends Operator<Nothing> { * @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 class GenericDataSinkBase<IN> extends Operator<Nothing> { */ @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 class GenericDataSinkBase<IN> extends Operator<Nothing> { */ @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 class GenericDataSinkBase<IN> extends Operator<Nothing> { */ @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 class GenericDataSinkBase<IN> extends Operator<Nothing> { @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 class GenericDataSinkBase<IN> extends Operator<Nothing> { format.configure(this.parameters); if(format instanceof RichOutputFormat){ - ((RichOutputFormat) format).setRuntimeContext(ctx); + ((RichOutputFormat<?>) format).setRuntimeContext(ctx); } format.open(0, 1); for (IN element : inputData) { http://git-wip-us.apache.org/repos/asf/flink/blob/760a0d9e/flink-core/src/main/java/org/apache/flink/api/common/operators/Keys.java ---------------------------------------------------------------------- diff --git a/flink-core/src/main/java/org/apache/flink/api/common/operators/Keys.java b/flink-core/src/main/java/org/apache/flink/api/common/operators/Keys.java index abe41af..0b771e8 100644 --- a/flink-core/src/main/java/org/apache/flink/api/common/operators/Keys.java +++ b/flink-core/src/main/java/org/apache/flink/api/common/operators/Keys.java @@ -18,10 +18,7 @@ package org.apache.flink.api.common.operators; -import java.util.ArrayList; -import java.util.List; - -import com.google.common.base.Joiner; +import org.apache.commons.lang3.StringUtils; import org.apache.flink.annotation.Internal; import org.apache.flink.api.common.InvalidProgramException; @@ -33,7 +30,11 @@ import org.apache.flink.api.java.functions.KeySelector; import org.apache.flink.api.java.typeutils.GenericTypeInfo; import org.apache.flink.api.java.typeutils.TypeExtractor; -import com.google.common.base.Preconditions; +import java.util.ArrayList; +import java.util.List; + +import static org.apache.flink.util.Preconditions.checkArgument; +import static org.apache.flink.util.Preconditions.checkNotNull; @Internal public abstract class Keys<T> { @@ -232,7 +233,8 @@ public abstract class Keys<T> { } else { rangeCheckFields(keyPositions, type.getArity() - 1); } - Preconditions.checkArgument(keyPositions.length > 0, "Grouping fields can not be empty at this point"); + + checkArgument(keyPositions.length > 0, "Grouping fields can not be empty at this point"); // extract key field types CompositeType<T> cType = (CompositeType<T>)type; @@ -266,7 +268,7 @@ public abstract class Keys<T> { * Create String-based (nested) field expression keys on a composite type. */ public ExpressionKeys(String[] keyExpressions, TypeInformation<T> type) { - Preconditions.checkNotNull(keyExpressions, "Field expression cannot be null."); + checkNotNull(keyExpressions, "Field expression cannot be null."); this.keyFields = new ArrayList<>(keyExpressions.length); @@ -375,8 +377,7 @@ public abstract class Keys<T> { @Override public String toString() { - Joiner join = Joiner.on('.'); - return "ExpressionKeys: " + join.join(keyFields); + return "ExpressionKeys: " + StringUtils.join(keyFields, '.'); } public static boolean isSortKey(int fieldPos, TypeInformation<?> type) { http://git-wip-us.apache.org/repos/asf/flink/blob/760a0d9e/flink-core/src/main/java/org/apache/flink/api/common/operators/base/GroupCombineOperatorBase.java ---------------------------------------------------------------------- diff --git a/flink-core/src/main/java/org/apache/flink/api/common/operators/base/GroupCombineOperatorBase.java b/flink-core/src/main/java/org/apache/flink/api/common/operators/base/GroupCombineOperatorBase.java index b660506..15a0f5a 100644 --- a/flink-core/src/main/java/org/apache/flink/api/common/operators/base/GroupCombineOperatorBase.java +++ b/flink-core/src/main/java/org/apache/flink/api/common/operators/base/GroupCombineOperatorBase.java @@ -18,8 +18,8 @@ package org.apache.flink.api.common.operators.base; -import com.google.common.base.Preconditions; import org.apache.commons.lang3.ArrayUtils; + import org.apache.flink.annotation.Internal; import org.apache.flink.api.common.ExecutionConfig; import org.apache.flink.api.common.InvalidProgramException; @@ -43,6 +43,8 @@ import java.util.Collections; import java.util.Comparator; import java.util.List; +import static org.apache.flink.util.Preconditions.checkArgument; + /** * Base operator for the combineGroup transformation. It receives the UDF GroupCombineFunction as an input. * This class is later processed by the compiler to generate the plan. @@ -109,7 +111,7 @@ public class GroupCombineOperatorBase<IN, OUT, FT extends GroupCombineFunction<I } if(sortColumns.length == 0) { // => all reduce. No comparator - Preconditions.checkArgument(sortOrderings.length == 0); + checkArgument(sortOrderings.length == 0); } else { final TypeComparator<IN> sortComparator = getTypeComparator(inputType, sortColumns, sortOrderings, executionConfig); http://git-wip-us.apache.org/repos/asf/flink/blob/760a0d9e/flink-core/src/main/java/org/apache/flink/api/common/operators/base/GroupReduceOperatorBase.java ---------------------------------------------------------------------- diff --git a/flink-core/src/main/java/org/apache/flink/api/common/operators/base/GroupReduceOperatorBase.java b/flink-core/src/main/java/org/apache/flink/api/common/operators/base/GroupReduceOperatorBase.java index 0cc0209..0794a77 100644 --- a/flink-core/src/main/java/org/apache/flink/api/common/operators/base/GroupReduceOperatorBase.java +++ b/flink-core/src/main/java/org/apache/flink/api/common/operators/base/GroupReduceOperatorBase.java @@ -41,13 +41,13 @@ import org.apache.flink.api.common.typeutils.CompositeType; import org.apache.flink.api.common.typeutils.TypeComparator; import org.apache.flink.api.common.typeutils.TypeSerializer; -import com.google.common.base.Preconditions; - import java.util.ArrayList; import java.util.Collections; import java.util.Comparator; import java.util.List; +import static org.apache.flink.util.Preconditions.checkArgument; + /** * @see org.apache.flink.api.common.functions.GroupReduceFunction */ @@ -181,7 +181,7 @@ public class GroupReduceOperatorBase<IN, OUT, FT extends GroupReduceFunction<IN, } if(sortColumns.length == 0) { // => all reduce. No comparator - Preconditions.checkArgument(sortOrderings.length == 0); + checkArgument(sortOrderings.length == 0); } else { final TypeComparator<IN> sortComparator = getTypeComparator(inputType, sortColumns, sortOrderings, executionConfig); Collections.sort(inputData, new Comparator<IN>() { http://git-wip-us.apache.org/repos/asf/flink/blob/760a0d9e/flink-core/src/main/java/org/apache/flink/api/common/operators/util/FieldList.java ---------------------------------------------------------------------- diff --git a/flink-core/src/main/java/org/apache/flink/api/common/operators/util/FieldList.java b/flink-core/src/main/java/org/apache/flink/api/common/operators/util/FieldList.java index c69c875..15a993c 100644 --- a/flink-core/src/main/java/org/apache/flink/api/common/operators/util/FieldList.java +++ b/flink-core/src/main/java/org/apache/flink/api/common/operators/util/FieldList.java @@ -16,16 +16,16 @@ * limitations under the License. */ - package org.apache.flink.api.common.operators.util; import java.util.ArrayList; import java.util.Collections; import java.util.List; -import com.google.common.base.Preconditions; import org.apache.flink.annotation.Internal; +import static org.apache.flink.util.Preconditions.checkNotNull; + /** * Immutable ordered list of fields IDs. */ @@ -44,7 +44,7 @@ public class FieldList extends FieldSet { } public FieldList(Integer fieldId) { - super(Collections.singletonList(Preconditions.checkNotNull(fieldId, "The fields ID must not be null."))); + super(Collections.singletonList(checkNotNull(fieldId, "The fields ID must not be null."))); } public FieldList(int... columnIndexes) { http://git-wip-us.apache.org/repos/asf/flink/blob/760a0d9e/flink-core/src/main/java/org/apache/flink/api/common/operators/util/UserCodeObjectWrapper.java ---------------------------------------------------------------------- diff --git a/flink-core/src/main/java/org/apache/flink/api/common/operators/util/UserCodeObjectWrapper.java b/flink-core/src/main/java/org/apache/flink/api/common/operators/util/UserCodeObjectWrapper.java index 890caac..5d75c95 100644 --- a/flink-core/src/main/java/org/apache/flink/api/common/operators/util/UserCodeObjectWrapper.java +++ b/flink-core/src/main/java/org/apache/flink/api/common/operators/util/UserCodeObjectWrapper.java @@ -27,7 +27,8 @@ import java.lang.reflect.Modifier; import org.apache.flink.annotation.Internal; import org.apache.flink.api.common.NonSerializableUserCodeException; -import com.google.common.base.Preconditions; +import static org.apache.flink.util.Preconditions.checkArgument; +import static org.apache.flink.util.Preconditions.checkNotNull; /** * This holds an actual object containing user defined code. @@ -39,8 +40,8 @@ public class UserCodeObjectWrapper<T> implements UserCodeWrapper<T> { private final T userCodeObject; public UserCodeObjectWrapper(T userCodeObject) { - Preconditions.checkNotNull(userCodeObject, "The user code object may not be null."); - Preconditions.checkArgument(userCodeObject instanceof Serializable, "User code object is not serializable: " + userCodeObject.getClass().getName()); + checkNotNull(userCodeObject, "The user code object may not be null."); + checkArgument(userCodeObject instanceof Serializable, "User code object is not serializable: " + userCodeObject.getClass().getName()); this.userCodeObject = userCodeObject; http://git-wip-us.apache.org/repos/asf/flink/blob/760a0d9e/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/BasicArrayTypeInfo.java ---------------------------------------------------------------------- diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/BasicArrayTypeInfo.java b/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/BasicArrayTypeInfo.java index 2c61fb2..25b2850 100644 --- a/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/BasicArrayTypeInfo.java +++ b/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/BasicArrayTypeInfo.java @@ -22,7 +22,6 @@ import java.util.HashMap; import java.util.Map; import java.util.Objects; -import com.google.common.base.Preconditions; import org.apache.flink.annotation.PublicEvolving; import org.apache.flink.annotation.Public; import org.apache.flink.api.common.ExecutionConfig; @@ -31,6 +30,8 @@ import org.apache.flink.api.common.typeutils.base.array.StringArraySerializer; import org.apache.flink.api.common.functions.InvalidTypesException; import org.apache.flink.api.common.typeutils.base.GenericArraySerializer; +import static org.apache.flink.util.Preconditions.checkNotNull; + @Public public final class BasicArrayTypeInfo<T, C> extends TypeInformation<T> { @@ -51,11 +52,10 @@ public final class BasicArrayTypeInfo<T, C> extends TypeInformation<T> { private final Class<T> arrayClass; private final TypeInformation<C> componentInfo; - - @SuppressWarnings("unchecked") + private BasicArrayTypeInfo(Class<T> arrayClass, BasicTypeInfo<C> componentInfo) { - this.arrayClass = Preconditions.checkNotNull(arrayClass); - this.componentInfo = Preconditions.checkNotNull(componentInfo); + this.arrayClass = checkNotNull(arrayClass); + this.componentInfo = checkNotNull(componentInfo); } // -------------------------------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/flink/blob/760a0d9e/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/BasicTypeInfo.java ---------------------------------------------------------------------- diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/BasicTypeInfo.java b/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/BasicTypeInfo.java index 4eb70c1..c1d5605 100644 --- a/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/BasicTypeInfo.java +++ b/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/BasicTypeInfo.java @@ -25,7 +25,6 @@ import java.util.HashMap; import java.util.Map; import java.util.Objects; -import com.google.common.base.Preconditions; import org.apache.flink.annotation.PublicEvolving; import org.apache.flink.annotation.Public; import org.apache.flink.api.common.ExecutionConfig; @@ -54,6 +53,7 @@ import org.apache.flink.api.common.typeutils.base.StringComparator; import org.apache.flink.api.common.typeutils.base.StringSerializer; import org.apache.flink.api.common.typeutils.base.VoidSerializer; +import static org.apache.flink.util.Preconditions.checkNotNull; /** * Type information for primitive types (int, long, double, byte, ...), String, Date, and Void. @@ -87,9 +87,9 @@ public class BasicTypeInfo<T> extends TypeInformation<T> implements AtomicType<T protected BasicTypeInfo(Class<T> clazz, Class<?>[] possibleCastTargetTypes, TypeSerializer<T> serializer, Class<? extends TypeComparator<T>> comparatorClass) { - this.clazz = Preconditions.checkNotNull(clazz); - this.possibleCastTargetTypes = Preconditions.checkNotNull(possibleCastTargetTypes); - this.serializer = Preconditions.checkNotNull(serializer); + this.clazz = checkNotNull(clazz); + this.possibleCastTargetTypes = checkNotNull(possibleCastTargetTypes); + this.serializer = checkNotNull(serializer); // comparator can be null as in VOID_TYPE_INFO this.comparatorClass = comparatorClass; } http://git-wip-us.apache.org/repos/asf/flink/blob/760a0d9e/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/FractionalTypeInfo.java ---------------------------------------------------------------------- diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/FractionalTypeInfo.java b/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/FractionalTypeInfo.java index aa22ac6..b4e53d4 100644 --- a/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/FractionalTypeInfo.java +++ b/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/FractionalTypeInfo.java @@ -18,13 +18,14 @@ package org.apache.flink.api.common.typeinfo; -import com.google.common.base.Preconditions; -import com.google.common.collect.Sets; import org.apache.flink.annotation.Public; import org.apache.flink.api.common.typeutils.TypeComparator; import org.apache.flink.api.common.typeutils.TypeSerializer; -import java.util.Set; +import java.util.Arrays; +import java.util.HashSet; + +import static org.apache.flink.util.Preconditions.checkArgument; /** * Type information for numeric fractional primitive types (double, float). @@ -34,15 +35,15 @@ public class FractionalTypeInfo<T> extends NumericTypeInfo<T> { private static final long serialVersionUID = 554334260950199994L; - private static final Set<Class<?>> fractionalTypes = Sets.<Class<?>>newHashSet( - Double.class, - Float.class - ); + private static final HashSet<Class<?>> fractionalTypes = new HashSet<Class<?>>( + Arrays.asList( + Double.class, + Float.class)); protected FractionalTypeInfo(Class<T> clazz, Class<?>[] possibleCastTargetTypes, TypeSerializer<T> serializer, Class<? extends TypeComparator<T>> comparatorClass) { super(clazz, possibleCastTargetTypes, serializer, comparatorClass); - Preconditions.checkArgument(fractionalTypes.contains(clazz), "The given class " + - clazz.getSimpleName() + " is not a fractional type."); + checkArgument(fractionalTypes.contains(clazz), + "The given class %s is not a fractional type.", clazz.getSimpleName()); } } http://git-wip-us.apache.org/repos/asf/flink/blob/760a0d9e/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/IntegerTypeInfo.java ---------------------------------------------------------------------- diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/IntegerTypeInfo.java b/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/IntegerTypeInfo.java index bff3ab7..02b416a 100644 --- a/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/IntegerTypeInfo.java +++ b/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/IntegerTypeInfo.java @@ -18,13 +18,14 @@ package org.apache.flink.api.common.typeinfo; -import com.google.common.base.Preconditions; -import com.google.common.collect.Sets; import org.apache.flink.annotation.Public; import org.apache.flink.api.common.typeutils.TypeComparator; import org.apache.flink.api.common.typeutils.TypeSerializer; -import java.util.Set; +import java.util.Arrays; +import java.util.HashSet; + +import static org.apache.flink.util.Preconditions.checkArgument; /** * Type information for numeric integer primitive types: int, long, byte, short, character. @@ -34,18 +35,18 @@ public class IntegerTypeInfo<T> extends NumericTypeInfo<T> { private static final long serialVersionUID = -8068827354966766955L; - private static final Set<Class<?>> integerTypes = Sets.<Class<?>>newHashSet( - Integer.class, - Long.class, - Byte.class, - Short.class, - Character.class - ); + private static final HashSet<Class<?>> integerTypes = new HashSet<Class<?>>( + Arrays.asList( + Integer.class, + Long.class, + Byte.class, + Short.class, + Character.class)); protected IntegerTypeInfo(Class<T> clazz, Class<?>[] possibleCastTargetTypes, TypeSerializer<T> serializer, Class<? extends TypeComparator<T>> comparatorClass) { super(clazz, possibleCastTargetTypes, serializer, comparatorClass); - Preconditions.checkArgument(integerTypes.contains(clazz), "The given class " + - clazz.getSimpleName() + " is not a integer type."); + checkArgument(integerTypes.contains(clazz), + "The given class %s is not a integer type.", clazz.getSimpleName()); } } http://git-wip-us.apache.org/repos/asf/flink/blob/760a0d9e/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/NumericTypeInfo.java ---------------------------------------------------------------------- diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/NumericTypeInfo.java b/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/NumericTypeInfo.java index 6969520..fea8abe 100644 --- a/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/NumericTypeInfo.java +++ b/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/NumericTypeInfo.java @@ -18,13 +18,14 @@ package org.apache.flink.api.common.typeinfo; -import com.google.common.base.Preconditions; -import com.google.common.collect.Sets; import org.apache.flink.annotation.Public; import org.apache.flink.api.common.typeutils.TypeComparator; import org.apache.flink.api.common.typeutils.TypeSerializer; -import java.util.Set; +import java.util.Arrays; +import java.util.HashSet; + +import static org.apache.flink.util.Preconditions.checkArgument; /** * Type information for numeric primitive types: int, long, double, byte, short, float, char. @@ -34,21 +35,21 @@ public abstract class NumericTypeInfo<T> extends BasicTypeInfo<T> { private static final long serialVersionUID = -5937777910658986986L; - private static final Set<Class<?>> numericalTypes = Sets.<Class<?>>newHashSet( - Integer.class, - Long.class, - Double.class, - Byte.class, - Short.class, - Float.class, - Character.class - ); + private static final HashSet<Class<?>> numericalTypes = new HashSet<Class<?>>( + Arrays.asList( + Integer.class, + Long.class, + Double.class, + Byte.class, + Short.class, + Float.class, + Character.class)); protected NumericTypeInfo(Class<T> clazz, Class<?>[] possibleCastTargetTypes, TypeSerializer<T> serializer, Class<? extends TypeComparator<T>> comparatorClass) { super(clazz, possibleCastTargetTypes, serializer, comparatorClass); - Preconditions.checkArgument(numericalTypes.contains(clazz), "The given class " + - clazz.getSimpleName() + " is not a numerical type."); + checkArgument(numericalTypes.contains(clazz), + "The given class %s is not a numerical type", clazz.getSimpleName()); } } http://git-wip-us.apache.org/repos/asf/flink/blob/760a0d9e/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/PrimitiveArrayTypeInfo.java ---------------------------------------------------------------------- diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/PrimitiveArrayTypeInfo.java b/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/PrimitiveArrayTypeInfo.java index 2c75458..1c6ce00 100644 --- a/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/PrimitiveArrayTypeInfo.java +++ b/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/PrimitiveArrayTypeInfo.java @@ -18,13 +18,8 @@ package org.apache.flink.api.common.typeinfo; -import java.util.HashMap; -import java.util.Map; -import java.util.Objects; - -import com.google.common.base.Preconditions; -import org.apache.flink.annotation.PublicEvolving; import org.apache.flink.annotation.Public; +import org.apache.flink.annotation.PublicEvolving; import org.apache.flink.api.common.ExecutionConfig; import org.apache.flink.api.common.functions.InvalidTypesException; import org.apache.flink.api.common.typeutils.TypeSerializer; @@ -46,6 +41,13 @@ import org.apache.flink.api.common.typeutils.base.array.PrimitiveArrayComparator import org.apache.flink.api.common.typeutils.base.array.ShortPrimitiveArrayComparator; import org.apache.flink.api.common.typeutils.base.array.ShortPrimitiveArraySerializer; +import java.util.HashMap; +import java.util.Map; +import java.util.Objects; + +import static org.apache.flink.util.Preconditions.checkArgument; +import static org.apache.flink.util.Preconditions.checkNotNull; + /** * A {@link TypeInformation} for arrays of primitive types (int, long, double, ...). * Supports the creation of dedicated efficient serializers for these types. @@ -83,11 +85,11 @@ public class PrimitiveArrayTypeInfo<T> extends TypeInformation<T> implements Ato * @param comparatorClass The class of the array comparator */ private PrimitiveArrayTypeInfo(Class<T> arrayClass, TypeSerializer<T> serializer, Class<? extends PrimitiveArrayComparator<T, ?>> comparatorClass) { - this.arrayClass = Preconditions.checkNotNull(arrayClass); - this.serializer = Preconditions.checkNotNull(serializer); - this.comparatorClass = Preconditions.checkNotNull(comparatorClass); + this.arrayClass = checkNotNull(arrayClass); + this.serializer = checkNotNull(serializer); + this.comparatorClass = checkNotNull(comparatorClass); - Preconditions.checkArgument( + checkArgument( arrayClass.isArray() && arrayClass.getComponentType().isPrimitive(), "Class must represent an array of primitives"); } http://git-wip-us.apache.org/repos/asf/flink/blob/760a0d9e/flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeType.java ---------------------------------------------------------------------- diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeType.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeType.java index 19b6eaf..4bf17ea 100644 --- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeType.java +++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeType.java @@ -22,13 +22,13 @@ import java.util.ArrayList; import java.util.List; import java.util.Objects; -import com.google.common.base.Preconditions; import org.apache.flink.annotation.PublicEvolving; import org.apache.flink.annotation.Public; import org.apache.flink.api.common.ExecutionConfig; import org.apache.flink.api.common.typeinfo.AtomicType; import org.apache.flink.api.common.typeinfo.TypeInformation; +import static org.apache.flink.util.Preconditions.checkNotNull; /** * Base type information class for Tuple and Pojo types @@ -44,7 +44,7 @@ public abstract class CompositeType<T> extends TypeInformation<T> { @PublicEvolving public CompositeType(Class<T> typeClass) { - this.typeClass = Preconditions.checkNotNull(typeClass); + this.typeClass = checkNotNull(typeClass); } /** http://git-wip-us.apache.org/repos/asf/flink/blob/760a0d9e/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/EnumSerializer.java ---------------------------------------------------------------------- diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/EnumSerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/EnumSerializer.java index c64e399..21a6ea0 100644 --- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/EnumSerializer.java +++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/EnumSerializer.java @@ -22,12 +22,13 @@ import java.io.IOException; import java.io.ObjectInputStream; import java.lang.reflect.Method; -import com.google.common.base.Preconditions; import org.apache.flink.annotation.Internal; import org.apache.flink.api.common.typeutils.TypeSerializer; import org.apache.flink.core.memory.DataInputView; import org.apache.flink.core.memory.DataOutputView; +import static org.apache.flink.util.Preconditions.checkNotNull; + @Internal public final class EnumSerializer<T extends Enum<T>> extends TypeSerializer<T> { @@ -38,7 +39,7 @@ public final class EnumSerializer<T extends Enum<T>> extends TypeSerializer<T> { private final Class<T> enumClass; public EnumSerializer(Class<T> enumClass) { - this.enumClass = Preconditions.checkNotNull(enumClass); + this.enumClass = checkNotNull(enumClass); this.values = createValues(enumClass); } http://git-wip-us.apache.org/repos/asf/flink/blob/760a0d9e/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/GenericArraySerializer.java ---------------------------------------------------------------------- diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/GenericArraySerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/GenericArraySerializer.java index eb6e708..f35d71b 100644 --- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/GenericArraySerializer.java +++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/GenericArraySerializer.java @@ -21,12 +21,13 @@ package org.apache.flink.api.common.typeutils.base; import java.io.IOException; import java.lang.reflect.Array; -import com.google.common.base.Preconditions; import org.apache.flink.annotation.Internal; import org.apache.flink.api.common.typeutils.TypeSerializer; import org.apache.flink.core.memory.DataInputView; import org.apache.flink.core.memory.DataOutputView; +import static org.apache.flink.util.Preconditions.checkNotNull; + /** * A serializer for arrays of objects. * @@ -45,8 +46,8 @@ public final class GenericArraySerializer<C> extends TypeSerializer<C[]> { public GenericArraySerializer(Class<C> componentClass, TypeSerializer<C> componentSerializer) { - this.componentClass = Preconditions.checkNotNull(componentClass); - this.componentSerializer = Preconditions.checkNotNull(componentSerializer); + this.componentClass = checkNotNull(componentClass); + this.componentSerializer = checkNotNull(componentSerializer); } @Override http://git-wip-us.apache.org/repos/asf/flink/blob/760a0d9e/flink-core/src/main/java/org/apache/flink/api/java/typeutils/EnumTypeInfo.java ---------------------------------------------------------------------- diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/EnumTypeInfo.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/EnumTypeInfo.java index aec3c1d..61d9ae4 100644 --- a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/EnumTypeInfo.java +++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/EnumTypeInfo.java @@ -18,7 +18,7 @@ package org.apache.flink.api.java.typeutils; -import com.google.common.base.Preconditions; +import org.apache.flink.annotation.Public; import org.apache.flink.annotation.PublicEvolving; import org.apache.flink.api.common.ExecutionConfig; import org.apache.flink.api.common.typeinfo.AtomicType; @@ -27,7 +27,8 @@ import org.apache.flink.api.common.typeutils.TypeComparator; import org.apache.flink.api.common.typeutils.TypeSerializer; import org.apache.flink.api.common.typeutils.base.EnumComparator; import org.apache.flink.api.common.typeutils.base.EnumSerializer; -import org.apache.flink.annotation.Public; + +import static org.apache.flink.util.Preconditions.checkNotNull; /** * A {@link TypeInformation} for java enumeration types. @@ -43,7 +44,7 @@ public class EnumTypeInfo<T extends Enum<T>> extends TypeInformation<T> implemen @PublicEvolving public EnumTypeInfo(Class<T> typeClass) { - Preconditions.checkNotNull(typeClass, "Enum type class must not be null."); + checkNotNull(typeClass, "Enum type class must not be null."); if (!Enum.class.isAssignableFrom(typeClass) ) { throw new IllegalArgumentException("EnumTypeInfo can only be used for subclasses of " + Enum.class.getName()); http://git-wip-us.apache.org/repos/asf/flink/blob/760a0d9e/flink-core/src/main/java/org/apache/flink/api/java/typeutils/GenericTypeInfo.java ---------------------------------------------------------------------- diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/GenericTypeInfo.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/GenericTypeInfo.java index 0cca8bd..bc4e87a 100644 --- a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/GenericTypeInfo.java +++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/GenericTypeInfo.java @@ -18,7 +18,6 @@ package org.apache.flink.api.java.typeutils; -import com.google.common.base.Preconditions; import org.apache.flink.annotation.PublicEvolving; import org.apache.flink.annotation.Public; import org.apache.flink.api.common.ExecutionConfig; @@ -29,6 +28,7 @@ import org.apache.flink.api.common.typeutils.TypeSerializer; import org.apache.flink.api.java.typeutils.runtime.GenericTypeComparator; import org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer; +import static org.apache.flink.util.Preconditions.checkNotNull; @Public public class GenericTypeInfo<T> extends TypeInformation<T> implements AtomicType<T> { @@ -39,7 +39,7 @@ public class GenericTypeInfo<T> extends TypeInformation<T> implements AtomicType @PublicEvolving public GenericTypeInfo(Class<T> typeClass) { - this.typeClass = Preconditions.checkNotNull(typeClass); + this.typeClass = checkNotNull(typeClass); } @Override http://git-wip-us.apache.org/repos/asf/flink/blob/760a0d9e/flink-core/src/main/java/org/apache/flink/api/java/typeutils/ObjectArrayTypeInfo.java ---------------------------------------------------------------------- diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/ObjectArrayTypeInfo.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/ObjectArrayTypeInfo.java index 1e8fbe2..2edb3b9 100644 --- a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/ObjectArrayTypeInfo.java +++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/ObjectArrayTypeInfo.java @@ -27,7 +27,8 @@ import org.apache.flink.api.common.typeinfo.TypeInformation; import org.apache.flink.api.common.typeutils.TypeSerializer; import org.apache.flink.api.common.typeutils.base.GenericArraySerializer; -import com.google.common.base.Preconditions; +import static org.apache.flink.util.Preconditions.checkNotNull; +import static org.apache.flink.util.Preconditions.checkArgument; @Public public class ObjectArrayTypeInfo<T, C> extends TypeInformation<T> { @@ -38,8 +39,8 @@ public class ObjectArrayTypeInfo<T, C> extends TypeInformation<T> { private final TypeInformation<C> componentInfo; private ObjectArrayTypeInfo(Class<T> arrayType, TypeInformation<C> componentInfo) { - this.arrayType = Preconditions.checkNotNull(arrayType); - this.componentInfo = Preconditions.checkNotNull(componentInfo); + this.arrayType = checkNotNull(arrayType); + this.componentInfo = checkNotNull(componentInfo); } // -------------------------------------------------------------------------------------------- @@ -128,9 +129,9 @@ public class ObjectArrayTypeInfo<T, C> extends TypeInformation<T> { @PublicEvolving public static <T, C> ObjectArrayTypeInfo<T, C> getInfoFor(Class<T> arrayClass, TypeInformation<C> componentInfo) { - Preconditions.checkNotNull(arrayClass); - Preconditions.checkNotNull(componentInfo); - Preconditions.checkArgument(arrayClass.isArray(), "Class " + arrayClass + " must be an array."); + checkNotNull(arrayClass); + checkNotNull(componentInfo); + checkArgument(arrayClass.isArray(), "Class " + arrayClass + " must be an array."); return new ObjectArrayTypeInfo<T, C>(arrayClass, componentInfo); } @@ -146,7 +147,7 @@ public class ObjectArrayTypeInfo<T, C> extends TypeInformation<T> { @SuppressWarnings("unchecked") @PublicEvolving public static <T, C> ObjectArrayTypeInfo<T, C> getInfoFor(TypeInformation<C> componentInfo) { - Preconditions.checkNotNull(componentInfo); + checkNotNull(componentInfo); return new ObjectArrayTypeInfo<T, C>( (Class<T>)Array.newInstance(componentInfo.getTypeClass(), 0).getClass(), http://git-wip-us.apache.org/repos/asf/flink/blob/760a0d9e/flink-core/src/main/java/org/apache/flink/api/java/typeutils/PojoField.java ---------------------------------------------------------------------- diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/PojoField.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/PojoField.java index c37fc77..026cfa6 100644 --- a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/PojoField.java +++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/PojoField.java @@ -25,10 +25,11 @@ import java.io.Serializable; import java.lang.reflect.Field; import java.util.Objects; -import com.google.common.base.Preconditions; import org.apache.flink.annotation.Internal; import org.apache.flink.api.common.typeinfo.TypeInformation; +import static org.apache.flink.util.Preconditions.checkNotNull; + /** * Represent a field definition for {@link PojoTypeInfo} type of objects. */ @@ -41,8 +42,8 @@ public class PojoField implements Serializable { private final TypeInformation<?> type; public PojoField(Field field, TypeInformation<?> type) { - this.field = Preconditions.checkNotNull(field); - this.type = Preconditions.checkNotNull(type); + this.field = checkNotNull(field); + this.type = checkNotNull(type); } public Field getField() { http://git-wip-us.apache.org/repos/asf/flink/blob/760a0d9e/flink-core/src/main/java/org/apache/flink/api/java/typeutils/PojoTypeInfo.java ---------------------------------------------------------------------- diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/PojoTypeInfo.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/PojoTypeInfo.java index be2a027..9c65263 100644 --- a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/PojoTypeInfo.java +++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/PojoTypeInfo.java @@ -18,19 +18,9 @@ package org.apache.flink.api.java.typeutils; -import java.lang.reflect.Field; -import java.lang.reflect.Modifier; -import java.util.ArrayList; -import java.util.Arrays; -import java.util.Comparator; -import java.util.List; -import java.util.regex.Matcher; -import java.util.regex.Pattern; - -import com.google.common.base.Preconditions; - -import org.apache.flink.annotation.PublicEvolving; +import org.apache.commons.lang3.StringUtils; import org.apache.flink.annotation.Public; +import org.apache.flink.annotation.PublicEvolving; import org.apache.flink.api.common.ExecutionConfig; import org.apache.flink.api.common.operators.Keys.ExpressionKeys; import org.apache.flink.api.common.typeinfo.TypeInformation; @@ -40,10 +30,20 @@ import org.apache.flink.api.common.typeutils.TypeSerializer; import org.apache.flink.api.java.typeutils.runtime.AvroSerializer; import org.apache.flink.api.java.typeutils.runtime.PojoComparator; import org.apache.flink.api.java.typeutils.runtime.PojoSerializer; - -import com.google.common.base.Joiner; import org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer; +import java.lang.reflect.Field; +import java.lang.reflect.Modifier; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Comparator; +import java.util.List; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +import static org.apache.flink.util.Preconditions.checkArgument; +import static org.apache.flink.util.Preconditions.checkState; + /** * TypeInformation for "Java Beans"-style types. Flink refers to them as POJOs, * since the conditions are slightly different from Java Beans. @@ -78,8 +78,8 @@ public class PojoTypeInfo<T> extends CompositeType<T> { public PojoTypeInfo(Class<T> typeClass, List<PojoField> fields) { super(typeClass); - Preconditions.checkArgument(Modifier.isPublic(typeClass.getModifiers()), - "POJO " + typeClass + " is not public"); + checkArgument(Modifier.isPublic(typeClass.getModifiers()), + "POJO %s is not public", typeClass); this.fields = fields.toArray(new PojoField[fields.size()]); @@ -350,7 +350,7 @@ public class PojoTypeInfo<T> extends CompositeType<T> { fieldStrings.add(field.getField().getName() + ": " + field.getTypeInformation().toString()); } return "PojoType<" + getTypeClass().getName() - + ", fields = [" + Joiner.on(", ").join(fieldStrings) + "]" + + ", fields = [" + StringUtils.join(fieldStrings, ", ") + "]" + ">"; } @@ -381,15 +381,15 @@ public class PojoTypeInfo<T> extends CompositeType<T> { @Override public TypeComparator<T> createTypeComparator(ExecutionConfig config) { - Preconditions.checkState( + checkState( keyFields.size() > 0, "No keys were defined for the PojoTypeComparatorBuilder."); - Preconditions.checkState( + checkState( fieldComparators.size() > 0, "No type comparators were defined for the PojoTypeComparatorBuilder."); - Preconditions.checkState( + checkState( keyFields.size() == fieldComparators.size(), "Number of key fields and field comparators is not equal."); http://git-wip-us.apache.org/repos/asf/flink/blob/760a0d9e/flink-core/src/main/java/org/apache/flink/api/java/typeutils/TupleTypeInfo.java ---------------------------------------------------------------------- diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/TupleTypeInfo.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/TupleTypeInfo.java index 9ecbe73..051ad0df 100644 --- a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/TupleTypeInfo.java +++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/TupleTypeInfo.java @@ -22,24 +22,26 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; -import com.google.common.base.Preconditions; -import com.google.common.primitives.Ints; import org.apache.flink.annotation.PublicEvolving; import org.apache.flink.annotation.Public; import org.apache.flink.api.common.ExecutionConfig; import org.apache.flink.api.common.functions.InvalidTypesException; import org.apache.flink.api.common.typeinfo.BasicTypeInfo; import org.apache.flink.api.common.typeinfo.TypeInformation; -//CHECKSTYLE.OFF: AvoidStarImport - Needed for TupleGenerator import org.apache.flink.api.common.typeutils.TypeComparator; import org.apache.flink.api.common.typeutils.TypeSerializer; -import org.apache.flink.api.java.tuple.*; import org.apache.flink.api.java.typeutils.runtime.Tuple0Serializer; -//CHECKSTYLE.ON: AvoidStarImport import org.apache.flink.api.java.typeutils.runtime.TupleComparator; import org.apache.flink.api.java.typeutils.runtime.TupleSerializer; import org.apache.flink.types.Value; +//CHECKSTYLE.OFF: AvoidStarImport - Needed for TupleGenerator +import org.apache.flink.api.java.tuple.*; +//CHECKSTYLE.ON: AvoidStarImport + +import static org.apache.flink.util.Preconditions.checkArgument; +import static org.apache.flink.util.Preconditions.checkState; + /** * A {@link TypeInformation} for the tuple types of the Java API. * @@ -62,7 +64,7 @@ public final class TupleTypeInfo<T extends Tuple> extends TupleTypeInfoBase<T> { public TupleTypeInfo(Class<T> tupleType, TypeInformation<?>... types) { super(tupleType, types); - Preconditions.checkArgument( + checkArgument( types.length <= Tuple.MAX_ARITY, "The tuple type exceeds the maximum supported arity."); @@ -131,24 +133,24 @@ public final class TupleTypeInfo<T extends Tuple> extends TupleTypeInfoBase<T> { @Override public TypeComparator<T> createTypeComparator(ExecutionConfig config) { - Preconditions.checkState( + checkState( fieldComparators.size() > 0, "No field comparators were defined for the TupleTypeComparatorBuilder." ); - Preconditions.checkState( + checkState( logicalKeyFields.size() > 0, "No key fields were defined for the TupleTypeComparatorBuilder." ); - Preconditions.checkState( + checkState( fieldComparators.size() == logicalKeyFields.size(), "The number of field comparators and key fields is not equal." ); final int maxKey = Collections.max(logicalKeyFields); - Preconditions.checkState( + checkState( maxKey >= 0, "The maximum key field must be greater or equal than 0." ); @@ -160,7 +162,7 @@ public final class TupleTypeInfo<T extends Tuple> extends TupleTypeInfoBase<T> { } return new TupleComparator<T>( - Ints.toArray(logicalKeyFields), + listToPrimitives(logicalKeyFields), fieldComparators.toArray(new TypeComparator[fieldComparators.size()]), fieldSerializers ); @@ -255,4 +257,12 @@ public final class TupleTypeInfo<T extends Tuple> extends TupleTypeInfoBase<T> { return (TupleTypeInfo<X>) new TupleTypeInfo<>(infos); } + + private static int[] listToPrimitives(ArrayList<Integer> ints) { + int[] result = new int[ints.size()]; + for (int i = 0; i < result.length; i++) { + result[i] = ints.get(i); + } + return result; + } } http://git-wip-us.apache.org/repos/asf/flink/blob/760a0d9e/flink-core/src/main/java/org/apache/flink/api/java/typeutils/TupleTypeInfoBase.java ---------------------------------------------------------------------- diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/TupleTypeInfoBase.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/TupleTypeInfoBase.java index 753eb66..807fd54 100644 --- a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/TupleTypeInfoBase.java +++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/TupleTypeInfoBase.java @@ -23,12 +23,12 @@ import java.util.List; import java.util.regex.Matcher; import java.util.regex.Pattern; -import com.google.common.base.Preconditions; - import org.apache.flink.api.common.operators.Keys.ExpressionKeys; import org.apache.flink.api.common.typeinfo.TypeInformation; import org.apache.flink.api.common.typeutils.CompositeType; +import static org.apache.flink.util.Preconditions.checkNotNull; + public abstract class TupleTypeInfoBase<T> extends CompositeType<T> { private static final long serialVersionUID = 1L; @@ -52,7 +52,7 @@ public abstract class TupleTypeInfoBase<T> extends CompositeType<T> { public TupleTypeInfoBase(Class<T> tupleType, TypeInformation<?>... types) { super(tupleType); - this.types = Preconditions.checkNotNull(types); + this.types = checkNotNull(types); int fieldCounter = 0; http://git-wip-us.apache.org/repos/asf/flink/blob/760a0d9e/flink-core/src/main/java/org/apache/flink/api/java/typeutils/TypeExtractor.java ---------------------------------------------------------------------- diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/TypeExtractor.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/TypeExtractor.java index 151f359..0469cc2 100644 --- a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/TypeExtractor.java +++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/TypeExtractor.java @@ -66,7 +66,7 @@ import org.apache.hadoop.io.Writable; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import com.google.common.base.Preconditions; +import static org.apache.flink.util.Preconditions.checkNotNull; /** * A utility for reflection analysis on classes, to determine the return type of implementations of transformation @@ -902,7 +902,7 @@ public class TypeExtractor { if (isClassType(originalType)) { originalTypeAsClass = typeToClass(originalType); } - Preconditions.checkNotNull(originalTypeAsClass, "originalType has an unexpected type"); + checkNotNull(originalTypeAsClass, "originalType has an unexpected type"); // check if the class we assumed to conform to the defining type so far is actually a pojo because the // original type contains additional fields. // check for additional fields. @@ -1466,7 +1466,7 @@ public class TypeExtractor { @SuppressWarnings({ "unchecked", "rawtypes" }) private <OUT,IN1,IN2> TypeInformation<OUT> privateGetForClass(Class<OUT> clazz, ArrayList<Type> typeHierarchy, ParameterizedType parameterizedType, TypeInformation<IN1> in1Type, TypeInformation<IN2> in2Type) { - Preconditions.checkNotNull(clazz); + checkNotNull(clazz); // Object is handled as generic type info if (clazz.equals(Object.class)) { @@ -1822,7 +1822,7 @@ public class TypeExtractor { @SuppressWarnings({ "unchecked", "rawtypes" }) private <X> TypeInformation<X> privateGetForObject(X value) { - Preconditions.checkNotNull(value); + checkNotNull(value); // check if we can extract the types from tuples, otherwise work with the class if (value instanceof Tuple) { http://git-wip-us.apache.org/repos/asf/flink/blob/760a0d9e/flink-core/src/main/java/org/apache/flink/api/java/typeutils/ValueTypeInfo.java ---------------------------------------------------------------------- diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/ValueTypeInfo.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/ValueTypeInfo.java index 7c173c0..495a324 100644 --- a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/ValueTypeInfo.java +++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/ValueTypeInfo.java @@ -18,9 +18,8 @@ package org.apache.flink.api.java.typeutils; -import com.google.common.base.Preconditions; -import org.apache.flink.annotation.PublicEvolving; import org.apache.flink.annotation.Public; +import org.apache.flink.annotation.PublicEvolving; import org.apache.flink.api.common.ExecutionConfig; import org.apache.flink.api.common.functions.InvalidTypesException; import org.apache.flink.api.common.typeinfo.AtomicType; @@ -44,6 +43,9 @@ import org.apache.flink.types.ShortValue; import org.apache.flink.types.StringValue; import org.apache.flink.types.Value; +import static org.apache.flink.util.Preconditions.checkArgument; +import static org.apache.flink.util.Preconditions.checkNotNull; + /** * Type information for data types that extend the {@link Value} interface. The value * interface allows types to define their custom serialization and deserialization routines. @@ -70,11 +72,11 @@ public class ValueTypeInfo<T extends Value> extends TypeInformation<T> implement @PublicEvolving public ValueTypeInfo(Class<T> type) { - this.type = Preconditions.checkNotNull(type); + this.type = checkNotNull(type); - Preconditions.checkArgument( + checkArgument( Value.class.isAssignableFrom(type) || type.equals(Value.class), - "ValueTypeInfo can only be used for subclasses of " + Value.class.getName()); + "ValueTypeInfo can only be used for subclasses of %s", Value.class.getName()); } @Override http://git-wip-us.apache.org/repos/asf/flink/blob/760a0d9e/flink-core/src/main/java/org/apache/flink/api/java/typeutils/WritableTypeInfo.java ---------------------------------------------------------------------- diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/WritableTypeInfo.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/WritableTypeInfo.java index 5e3b2bc..7ca7a91 100644 --- a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/WritableTypeInfo.java +++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/WritableTypeInfo.java @@ -18,7 +18,6 @@ package org.apache.flink.api.java.typeutils; -import com.google.common.base.Preconditions; import org.apache.flink.annotation.PublicEvolving; import org.apache.flink.annotation.Public; import org.apache.flink.api.common.ExecutionConfig; @@ -29,8 +28,12 @@ import org.apache.flink.api.common.typeutils.TypeComparator; import org.apache.flink.api.common.typeutils.TypeSerializer; import org.apache.flink.api.java.typeutils.runtime.WritableComparator; import org.apache.flink.api.java.typeutils.runtime.WritableSerializer; + import org.apache.hadoop.io.Writable; +import static org.apache.flink.util.Preconditions.checkNotNull; +import static org.apache.flink.util.Preconditions.checkArgument; + /** * Type information for data types that extend Hadoop's {@link Writable} interface. The Writable * interface defines the serialization and deserialization routines for the data type. @@ -46,11 +49,11 @@ public class WritableTypeInfo<T extends Writable> extends TypeInformation<T> imp @PublicEvolving public WritableTypeInfo(Class<T> typeClass) { - this.typeClass = Preconditions.checkNotNull(typeClass); + this.typeClass = checkNotNull(typeClass); - Preconditions.checkArgument( + checkArgument( Writable.class.isAssignableFrom(typeClass) && !typeClass.equals(Writable.class), - "WritableTypeInfo can only be used for subclasses of " + Writable.class.getName()); + "WritableTypeInfo can only be used for subclasses of %s", Writable.class.getName()); } @SuppressWarnings({ "rawtypes", "unchecked" }) http://git-wip-us.apache.org/repos/asf/flink/blob/760a0d9e/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/AvroSerializer.java ---------------------------------------------------------------------- diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/AvroSerializer.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/AvroSerializer.java index bc04367..4c2a7f9 100644 --- a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/AvroSerializer.java +++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/AvroSerializer.java @@ -20,11 +20,11 @@ package org.apache.flink.api.java.typeutils.runtime; import java.io.IOException; -import com.google.common.base.Preconditions; import org.apache.avro.generic.GenericData; import org.apache.avro.reflect.ReflectDatumReader; import org.apache.avro.reflect.ReflectDatumWriter; import org.apache.avro.util.Utf8; + import org.apache.flink.api.common.typeutils.TypeSerializer; import org.apache.flink.api.java.typeutils.runtime.kryo.Serializers; import org.apache.flink.core.memory.DataInputView; @@ -34,6 +34,7 @@ import org.apache.flink.util.InstantiationUtil; import com.esotericsoftware.kryo.Kryo; import org.objenesis.strategy.StdInstantiatorStrategy; +import static org.apache.flink.util.Preconditions.checkNotNull; /** * General purpose serialization. Currently using Apache Avro's Reflect-serializers for serialization and @@ -66,8 +67,8 @@ public final class AvroSerializer<T> extends TypeSerializer<T> { } public AvroSerializer(Class<T> type, Class<? extends T> typeToInstantiate) { - this.type = Preconditions.checkNotNull(type); - this.typeToInstantiate = Preconditions.checkNotNull(typeToInstantiate); + this.type = checkNotNull(type); + this.typeToInstantiate = checkNotNull(typeToInstantiate); InstantiationUtil.checkForInstantiation(typeToInstantiate); } http://git-wip-us.apache.org/repos/asf/flink/blob/760a0d9e/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/CopyableValueSerializer.java ---------------------------------------------------------------------- diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/CopyableValueSerializer.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/CopyableValueSerializer.java index 9e46f27..f30a767 100644 --- a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/CopyableValueSerializer.java +++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/CopyableValueSerializer.java @@ -20,13 +20,13 @@ package org.apache.flink.api.java.typeutils.runtime; import java.io.IOException; -import com.google.common.base.Preconditions; import org.apache.flink.api.common.typeutils.TypeSerializer; import org.apache.flink.core.memory.DataInputView; import org.apache.flink.core.memory.DataOutputView; import org.apache.flink.types.CopyableValue; import org.apache.flink.util.InstantiationUtil; +import static org.apache.flink.util.Preconditions.checkNotNull; public class CopyableValueSerializer<T extends CopyableValue<T>> extends TypeSerializer<T> { @@ -39,7 +39,7 @@ public class CopyableValueSerializer<T extends CopyableValue<T>> extends TypeSer public CopyableValueSerializer(Class<T> valueClass) { - this.valueClass = Preconditions.checkNotNull(valueClass); + this.valueClass = checkNotNull(valueClass); } @Override http://git-wip-us.apache.org/repos/asf/flink/blob/760a0d9e/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializer.java ---------------------------------------------------------------------- diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializer.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializer.java index de24956..9958540 100644 --- a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializer.java +++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializer.java @@ -32,7 +32,6 @@ import java.util.List; import java.util.Map; import java.util.Objects; -import com.google.common.base.Preconditions; import org.apache.flink.api.common.ExecutionConfig; import org.apache.flink.api.common.typeinfo.TypeInformation; import org.apache.flink.api.common.typeutils.TypeSerializer; @@ -40,6 +39,7 @@ import org.apache.flink.api.java.typeutils.TypeExtractor; import org.apache.flink.core.memory.DataInputView; import org.apache.flink.core.memory.DataOutputView; +import static org.apache.flink.util.Preconditions.checkNotNull; public final class PojoSerializer<T> extends TypeSerializer<T> { @@ -75,11 +75,11 @@ public final class PojoSerializer<T> extends TypeSerializer<T> { Field[] fields, ExecutionConfig executionConfig) { - this.clazz = Preconditions.checkNotNull(clazz); - this.fieldSerializers = (TypeSerializer<Object>[]) Preconditions.checkNotNull(fieldSerializers); - this.fields = Preconditions.checkNotNull(fields); + this.clazz = checkNotNull(clazz); + this.fieldSerializers = (TypeSerializer<Object>[]) checkNotNull(fieldSerializers); + this.fields = checkNotNull(fields); this.numFields = fieldSerializers.length; - this.executionConfig = Preconditions.checkNotNull(executionConfig); + this.executionConfig = checkNotNull(executionConfig); LinkedHashSet<Class<?>> registeredPojoTypes = executionConfig.getRegisteredPojoTypes(); http://git-wip-us.apache.org/repos/asf/flink/blob/760a0d9e/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/TupleSerializerBase.java ---------------------------------------------------------------------- diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/TupleSerializerBase.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/TupleSerializerBase.java index 5b5d462..5a93cc5 100644 --- a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/TupleSerializerBase.java +++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/TupleSerializerBase.java @@ -18,7 +18,6 @@ package org.apache.flink.api.java.typeutils.runtime; -import com.google.common.base.Preconditions; import org.apache.flink.api.common.typeutils.TypeSerializer; import org.apache.flink.core.memory.DataInputView; import org.apache.flink.core.memory.DataOutputView; @@ -27,6 +26,7 @@ import java.io.IOException; import java.util.Arrays; import java.util.Objects; +import static org.apache.flink.util.Preconditions.checkNotNull; public abstract class TupleSerializerBase<T> extends TypeSerializer<T> { @@ -42,8 +42,8 @@ public abstract class TupleSerializerBase<T> extends TypeSerializer<T> { @SuppressWarnings("unchecked") public TupleSerializerBase(Class<T> tupleClass, TypeSerializer<?>[] fieldSerializers) { - this.tupleClass = Preconditions.checkNotNull(tupleClass); - this.fieldSerializers = (TypeSerializer<Object>[]) Preconditions.checkNotNull(fieldSerializers); + this.tupleClass = checkNotNull(tupleClass); + this.fieldSerializers = (TypeSerializer<Object>[]) checkNotNull(fieldSerializers); this.arity = fieldSerializers.length; } http://git-wip-us.apache.org/repos/asf/flink/blob/760a0d9e/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/ValueSerializer.java ---------------------------------------------------------------------- diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/ValueSerializer.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/ValueSerializer.java index 9329866..73dc0fc 100644 --- a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/ValueSerializer.java +++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/ValueSerializer.java @@ -20,7 +20,6 @@ package org.apache.flink.api.java.typeutils.runtime; import java.io.IOException; -import com.google.common.base.Preconditions; import org.apache.flink.api.common.typeutils.TypeSerializer; import org.apache.flink.core.memory.DataInputView; import org.apache.flink.core.memory.DataOutputView; @@ -30,6 +29,8 @@ import org.apache.flink.util.InstantiationUtil; import com.esotericsoftware.kryo.Kryo; import org.objenesis.strategy.StdInstantiatorStrategy; +import static org.apache.flink.util.Preconditions.checkNotNull; + /** * Serializer for {@link Value} types. Uses the value's serialization methods, and uses * Kryo for deep object copies. @@ -49,7 +50,7 @@ public class ValueSerializer<T extends Value> extends TypeSerializer<T> { // -------------------------------------------------------------------------------------------- public ValueSerializer(Class<T> type) { - this.type = Preconditions.checkNotNull(type); + this.type = checkNotNull(type); } // -------------------------------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/flink/blob/760a0d9e/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializer.java ---------------------------------------------------------------------- diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializer.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializer.java index 276ffc4..d5c2f67 100644 --- a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializer.java +++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializer.java @@ -25,9 +25,9 @@ import com.esotericsoftware.kryo.factories.ReflectionSerializerFactory; import com.esotericsoftware.kryo.io.Input; import com.esotericsoftware.kryo.io.Output; import com.esotericsoftware.kryo.serializers.JavaSerializer; -import com.google.common.base.Preconditions; import org.apache.avro.generic.GenericData; + import org.apache.flink.api.common.ExecutionConfig; import org.apache.flink.api.common.typeutils.TypeSerializer; import org.apache.flink.api.java.typeutils.runtime.DataInputViewStream; @@ -36,7 +36,9 @@ import org.apache.flink.api.java.typeutils.runtime.NoFetchingInput; import org.apache.flink.api.java.typeutils.runtime.kryo.Serializers.SpecificInstanceCollectionSerializerForArrayList; import org.apache.flink.core.memory.DataInputView; import org.apache.flink.core.memory.DataOutputView; + import org.objenesis.strategy.StdInstantiatorStrategy; + import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -52,6 +54,8 @@ import java.util.LinkedHashSet; import java.util.Map; import java.util.Objects; +import static org.apache.flink.util.Preconditions.checkNotNull; + /** * A type serializer that serializes its type using the Kryo serialization * framework (https://github.com/EsotericSoftware/kryo). @@ -92,7 +96,7 @@ public class KryoSerializer<T> extends TypeSerializer<T> { // ------------------------------------------------------------------------ public KryoSerializer(Class<T> type, ExecutionConfig executionConfig){ - this.type = Preconditions.checkNotNull(type); + this.type = checkNotNull(type); this.defaultSerializers = executionConfig.getDefaultKryoSerializers(); this.defaultSerializerClasses = executionConfig.getDefaultKryoSerializerClasses();
