[FLINK-8105] Remove "unnecessary 'null' check before 'instanceof' exp…

…ression"

This closes apache#5034

flink-connectors/flink-hadoop-compatibility/src/main/java/org/apache/flink/api/java/hadoop/mapred/HadoopInputFormatBase.java

@@ -121,7 +121,7 @@ public BaseStatistics getStatistics(BaseStatistics cachedStats) throws IOExcepti
  return null;
  }
 
- final FileBaseStatistics cachedFileStats = (cachedStats != null && cachedStats instanceof FileBaseStatistics) ?
+ final FileBaseStatistics cachedFileStats = (cachedStats instanceof FileBaseStatistics) ?
  (FileBaseStatistics) cachedStats : null;
 
  try {

flink-connectors/flink-hadoop-compatibility/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopInputFormatBase.java

@@ -119,7 +119,7 @@ public BaseStatistics getStatistics(BaseStatistics cachedStats) throws IOExcepti
 
  JobContext jobContext = new JobContextImpl(configuration, null);
 
- final FileBaseStatistics cachedFileStats = (cachedStats != null && cachedStats instanceof FileBaseStatistics) ?
+ final FileBaseStatistics cachedFileStats = (cachedStats instanceof FileBaseStatistics) ?
  (FileBaseStatistics) cachedStats : null;
 
  try {

flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/api/FlinkLocalCluster.java

@@ -73,7 +73,7 @@ public void submitTopologyWithOpts(final String topologyName, final Map conf, fi
  boolean submitBlocking = false;
  if (conf != null) {
  Object blockingFlag = conf.get(SUBMIT_BLOCKING);
- if (blockingFlag != null && blockingFlag instanceof Boolean) {
+ if (blockingFlag instanceof Boolean) {
  submitBlocking = ((Boolean) blockingFlag).booleanValue();
  }
  }

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

@@ -169,7 +169,7 @@ protected List<FileStatus> getFiles() throws IOException {
  @Override
  public SequentialStatistics getStatistics(BaseStatistics cachedStats) {
 
- final FileBaseStatistics cachedFileStats = (cachedStats != null && cachedStats instanceof FileBaseStatistics) ?
+ final FileBaseStatistics cachedFileStats = cachedStats instanceof FileBaseStatistics ?
  (FileBaseStatistics) cachedStats : null;
 
  try {

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

@@ -345,7 +345,7 @@ public void configure(Configuration parameters) {
  @Override
  public FileBaseStatistics getStatistics(BaseStatistics cachedStats) throws IOException {
 
- final FileBaseStatistics cachedFileStats = (cachedStats != null && cachedStats instanceof FileBaseStatistics) ?
+ final FileBaseStatistics cachedFileStats = cachedStats instanceof FileBaseStatistics ?
  (FileBaseStatistics) cachedStats : null;
 
  // store properties

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

@@ -379,7 +379,7 @@ public void configure(Configuration parameters) {
  @Override
  public FileBaseStatistics getStatistics(BaseStatistics cachedStats) throws IOException {
 
- final FileBaseStatistics cachedFileStats = (cachedStats != null && cachedStats instanceof FileBaseStatistics) ?
+ final FileBaseStatistics cachedFileStats = cachedStats instanceof FileBaseStatistics ?
  (FileBaseStatistics) cachedStats : null;
 
  try {

flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/RuntimeSerializerFactory.java

@@ -114,7 +114,7 @@ public int hashCode() {
 
  @Override
  public boolean equals(Object obj) {
- if (obj != null && obj instanceof RuntimeSerializerFactory) {
+ if (obj instanceof RuntimeSerializerFactory) {
  RuntimeSerializerFactory<?> other = (RuntimeSerializerFactory<?>) obj;
 
  return this.clazz == other.clazz &&

flink-core/src/main/java/org/apache/flink/core/fs/FileInputSplit.java

@@ -104,7 +104,7 @@ public boolean equals(Object obj) {
  if (obj == this) {
  return true;
  }
- else if (obj != null && obj instanceof FileInputSplit && super.equals(obj)) {
+ else if (obj instanceof FileInputSplit && super.equals(obj)) {
  FileInputSplit other = (FileInputSplit) obj;
 
  return this.start == other.start &&

flink-core/src/main/java/org/apache/flink/core/io/GenericInputSplit.java

@@ -67,7 +67,7 @@ public int hashCode() {
 
  @Override
  public boolean equals(Object obj) {
- if (obj != null && obj instanceof GenericInputSplit) {
+ if (obj instanceof GenericInputSplit) {
  GenericInputSplit other = (GenericInputSplit) obj;
  return this.partitionNumber == other.partitionNumber &&
  this.totalNumberOfPartitions == other.totalNumberOfPartitions;

flink-core/src/main/java/org/apache/flink/core/io/LocatableInputSplit.java

@@ -90,7 +90,7 @@ public boolean equals(Object obj) {
  if (obj == this) {
  return true;
  }
- else if (obj != null && obj instanceof LocatableInputSplit) {
+ else if (obj instanceof LocatableInputSplit) {
  LocatableInputSplit other = (LocatableInputSplit) obj;
  return other.splitNumber == this.splitNumber && Arrays.deepEquals(other.hostnames, this.hostnames);
  }

flink-java/src/main/java/org/apache/flink/api/java/sca/UdfAnalyzer.java

@@ -331,7 +331,7 @@ else if (returnValue.isInput() && !returnValue.isGrouped()) {
  while (cause != null && !(cause instanceof CodeErrorException)) {
  cause = cause.getCause();
  }
- if ((cause != null && cause instanceof CodeErrorException) || e instanceof CodeErrorException) {
+ if (cause instanceof CodeErrorException || e instanceof CodeErrorException) {
  throw new CodeErrorException("Function code contains obvious errors. " +
  "If you think the code analysis is wrong at this point you can " +
  "disable the entire code analyzer in ExecutionConfig or add" +

flink-java/src/main/java/org/apache/flink/api/java/sca/UdfAnalyzerUtils.java

@@ -155,7 +155,7 @@ public static Object[] findMethodNode(String internalClassName, String name, Str
  }
 
  public static boolean isTagged(Value value) {
- return value != null && value instanceof TaggedValue;
+ return value instanceof TaggedValue;
  }
 
  public static TaggedValue tagged(Value value) {

flink-java/src/test/java/org/apache/flink/api/java/io/CollectionInputFormatTest.java

@@ -70,7 +70,7 @@ public int getId() {
 
  @Override
  public boolean equals(Object obj) {
- if (obj != null && obj instanceof ElementType) {
+ if (obj instanceof ElementType) {
  ElementType et = (ElementType) obj;
  return et.getId() == this.getId();
  } else {

flink-optimizer/src/main/java/org/apache/flink/optimizer/dataproperties/GlobalProperties.java

@@ -456,7 +456,7 @@ public int hashCode() {
 
  @Override
  public boolean equals(Object obj) {
- if (obj != null && obj instanceof GlobalProperties) {
+ if (obj instanceof GlobalProperties) {
  final GlobalProperties other = (GlobalProperties) obj;
  return (this.partitioning == other.partitioning)
  && (this.ordering == other.ordering || (this.ordering != null && this.ordering.equals(other.ordering)))

flink-optimizer/src/main/java/org/apache/flink/optimizer/dataproperties/InterestingProperties.java

@@ -148,7 +148,7 @@ public int hashCode() {
 
  @Override
  public boolean equals(Object obj) {
- if (obj != null && obj instanceof InterestingProperties) {
+ if (obj instanceof InterestingProperties) {
  InterestingProperties other = (InterestingProperties) obj;
  return this.globalProps.equals(other.globalProps) &&
  this.localProps.equals(other.localProps);

flink-optimizer/src/main/java/org/apache/flink/optimizer/dataproperties/RequestedGlobalProperties.java

@@ -448,7 +448,7 @@ public int hashCode() {
 
  @Override
  public boolean equals(Object obj) {
- if (obj != null && obj instanceof RequestedGlobalProperties) {
+ if (obj instanceof RequestedGlobalProperties) {
  RequestedGlobalProperties other = (RequestedGlobalProperties) obj;
  return (ordering == other.getOrdering() || (ordering != null && ordering.equals(other.getOrdering())))
  && (partitioning == other.getPartitioning())

flink-optimizer/src/main/java/org/apache/flink/optimizer/plantranslate/JobGraphGenerator.java

@@ -810,16 +810,15 @@ private JobVertex createSingleInputVertex(SingleInputPlanNode node) throws Compi
 
  // cannot chain the nodes that produce the next workset or the next solution set, if they are not the
  // in a tail 
- if (this.currentIteration != null && this.currentIteration instanceof WorksetIterationPlanNode &&
- node.getOutgoingChannels().size() > 0)
+ if (this.currentIteration instanceof WorksetIterationPlanNode && node.getOutgoingChannels().size() > 0)
  {
  WorksetIterationPlanNode wspn = (WorksetIterationPlanNode) this.currentIteration;
  if (wspn.getSolutionSetDeltaPlanNode() == pred || wspn.getNextWorkSetPlanNode() == pred) {
  chaining = false;
  }
  }
  // cannot chain the nodes that produce the next workset in a bulk iteration if a termination criterion follows
- if (this.currentIteration != null && this.currentIteration instanceof BulkIterationPlanNode)
+ if (this.currentIteration instanceof BulkIterationPlanNode)
  {
  BulkIterationPlanNode wspn = (BulkIterationPlanNode) this.currentIteration;
  if (node == wspn.getRootOfTerminationCriterion() && wspn.getRootOfStepFunction() == pred){

flink-optimizer/src/main/java/org/apache/flink/optimizer/traversals/PlanFinalizer.java

@@ -147,7 +147,7 @@ else if (visitable instanceof BulkPartialSolutionPlanNode) {
  final IterationPlanNode iteration = this.stackOfIterationNodes.peekLast();
 
  // sanity check!
- if (iteration == null || !(iteration instanceof BulkIterationPlanNode)) {
+ if (!(iteration instanceof BulkIterationPlanNode)) {
  throw new CompilerException("Bug: Error finalizing the plan. " +
  "Cannot associate the node for a partial solutions with its containing iteration.");
  }
@@ -159,7 +159,7 @@ else if (visitable instanceof WorksetPlanNode) {
  final IterationPlanNode iteration = this.stackOfIterationNodes.peekLast();
 
  // sanity check!
- if (iteration == null || !(iteration instanceof WorksetIterationPlanNode)) {
+ if (!(iteration instanceof WorksetIterationPlanNode)) {
  throw new CompilerException("Bug: Error finalizing the plan. " +
  "Cannot associate the node for a partial solutions with its containing iteration.");
  }
@@ -171,7 +171,7 @@ else if (visitable instanceof SolutionSetPlanNode) {
  final IterationPlanNode iteration = this.stackOfIterationNodes.peekLast();
 
  // sanity check!
- if (iteration == null || !(iteration instanceof WorksetIterationPlanNode)) {
+ if (!(iteration instanceof WorksetIterationPlanNode)) {
  throw new CompilerException("Bug: Error finalizing the plan. " +
  "Cannot associate the node for a partial solutions with its containing iteration.");
  }

flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerRegistrationTest.java

@@ -545,7 +545,7 @@ protected void run() {
 
  // we might also receive RegisterTaskManager and Heartbeat messages which
  // are queued up in the testing actor's mailbox
- while(message == null || !(message instanceof Terminated)) {
+ while(!(message instanceof Terminated)) {
  message = receiveOne(timeout);
  }
 

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/TimestampedFileInputSplit.java

@@ -119,7 +119,7 @@ public int compareTo(TimestampedFileInputSplit o) {
  public boolean equals(Object o) {
  if (this == o) {
  return true;
- } else if (o != null && o instanceof TimestampedFileInputSplit && super.equals(o)) {
+ } else if (o instanceof TimestampedFileInputSplit && super.equals(o)) {
  TimestampedFileInputSplit that = (TimestampedFileInputSplit) o;
  return this.modificationTime == that.modificationTime;
  }

flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericWriteAheadSink.java

@@ -303,7 +303,7 @@ public int compareTo(PendingCheckpoint o) {
 
  @Override
  public boolean equals(Object o) {
- if (o == null || !(o instanceof GenericWriteAheadSink.PendingCheckpoint)) {
+ if (!(o instanceof GenericWriteAheadSink.PendingCheckpoint)) {
  return false;
  }
  PendingCheckpoint other = (PendingCheckpoint) o;