[FLINK-10208][build] Bump mockito to 2.21.0 / powermock to 2.0.0-beta.5

flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendSavepointTest.java

@@ -47,8 +47,8 @@
 import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
+import static org.mockito.ArgumentMatchers.nullable;
 import static org.mockito.Matchers.any;
-import static org.mockito.Matchers.anyString;
 import static org.mockito.Matchers.eq;
 import static org.mockito.Matchers.isNull;
 import static org.mockito.Mockito.mock;
@@ -315,7 +315,7 @@ private static void restoreStdOutAndStdErr() {
  private static ClusterClient<String> createClusterClient(String expectedResponse) throws Exception {
  final ClusterClient<String> clusterClient = mock(ClusterClient.class);
 
- when(clusterClient.triggerSavepoint(any(JobID.class), anyString()))
+ when(clusterClient.triggerSavepoint(any(JobID.class), nullable(String.class)))
  .thenReturn(CompletableFuture.completedFuture(expectedResponse));
 
  return clusterClient;
@@ -324,7 +324,7 @@ private static ClusterClient<String> createClusterClient(String expectedResponse
  private static ClusterClient<String> createFailingClusterClient(Exception expectedException) throws Exception {
  final ClusterClient<String> clusterClient = mock(ClusterClient.class);
 
- when(clusterClient.triggerSavepoint(any(JobID.class), anyString()))
+ when(clusterClient.triggerSavepoint(any(JobID.class), nullable(String.class)))
  .thenReturn(FutureUtils.completedExceptionally(expectedException));
 
  return clusterClient;

flink-connectors/flink-connector-elasticsearch-base/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchSinkBaseTest.java

@@ -483,7 +483,7 @@ protected BulkProcessor buildBulkProcessor(final BulkProcessor.Listener listener
  @Override
  public Object answer(InvocationOnMock invocationOnMock) throws Throwable {
  // intercept the request and add it to our mock bulk request
- nextBulkRequest.add(invocationOnMock.getArgumentAt(0, IndexRequest.class));
+ nextBulkRequest.add((IndexRequest) invocationOnMock.getArgument(0));
 
  return null;
  }

flink-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/internal/KafkaConsumerThreadTest.java

@@ -797,7 +797,7 @@ public Object answer(InvocationOnMock invocationOnMock) throws Throwable {
  when(mockConsumer.position(any(TopicPartition.class))).thenAnswer(new Answer<Object>() {
  @Override
  public Object answer(InvocationOnMock invocationOnMock) throws Throwable {
- return mockConsumerAssignmentAndPosition.get(invocationOnMock.getArgumentAt(0, TopicPartition.class));
+ return mockConsumerAssignmentAndPosition.get(invocationOnMock.getArgument(0));
  }
  });
  } else {
@@ -809,7 +809,7 @@ public Object answer(InvocationOnMock invocationOnMock) throws Throwable {
  public Object answer(InvocationOnMock invocationOnMock) throws Throwable {
  mockConsumerAssignmentAndPosition.clear();
 
- List<TopicPartition> assignedPartitions = invocationOnMock.getArgumentAt(0, List.class);
+ List<TopicPartition> assignedPartitions = invocationOnMock.getArgument(0);
  for (TopicPartition assigned : assignedPartitions) {
  mockConsumerAssignmentAndPosition.put(assigned, null);
  }
@@ -820,8 +820,8 @@ public Object answer(InvocationOnMock invocationOnMock) throws Throwable {
  doAnswer(new Answer() {
  @Override
  public Object answer(InvocationOnMock invocationOnMock) throws Throwable {
- TopicPartition partition = invocationOnMock.getArgumentAt(0, TopicPartition.class);
- long position = invocationOnMock.getArgumentAt(1, long.class);
+ TopicPartition partition = invocationOnMock.getArgument(0);
+ long position = invocationOnMock.getArgument(1);
 
  if (!mockConsumerAssignmentAndPosition.containsKey(partition)) {
  throw new Exception("the current mock assignment does not contain partition " + partition);
@@ -835,7 +835,7 @@ public Object answer(InvocationOnMock invocationOnMock) throws Throwable {
  doAnswer(new Answer() {
  @Override
  public Object answer(InvocationOnMock invocationOnMock) throws Throwable {
- TopicPartition partition = invocationOnMock.getArgumentAt(0, TopicPartition.class);
+ TopicPartition partition = invocationOnMock.getArgument(0);
 
  if (!mockConsumerAssignmentAndPosition.containsKey(partition)) {
  throw new Exception("the current mock assignment does not contain partition " + partition);
@@ -854,7 +854,7 @@ public Object answer(InvocationOnMock invocationOnMock) throws Throwable {
  doAnswer(new Answer() {
  @Override
  public Object answer(InvocationOnMock invocationOnMock) throws Throwable {
- TopicPartition partition = invocationOnMock.getArgumentAt(0, TopicPartition.class);
+ TopicPartition partition = invocationOnMock.getArgument(0);
 
  if (!mockConsumerAssignmentAndPosition.containsKey(partition)) {
  throw new Exception("the current mock assignment does not contain partition " + partition);

flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducerBaseTest.java

@@ -352,7 +352,7 @@ private static class DummyFlinkKafkaProducer<T> extends FlinkKafkaProducerBase<T
  when(mockProducer.send(any(ProducerRecord.class), any(Callback.class))).thenAnswer(new Answer<Object>() {
  @Override
  public Object answer(InvocationOnMock invocationOnMock) throws Throwable {
- pendingCallbacks.add(invocationOnMock.getArgumentAt(1, Callback.class));
+ pendingCallbacks.add(invocationOnMock.getArgument(1));
  return null;
  }
  });

flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/FlinkKinesisConsumerTest.java

@@ -28,6 +28,7 @@
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.api.java.typeutils.runtime.PojoSerializer;
 import org.apache.flink.configuration.Configuration;
+import org.apache.flink.mock.Whitebox;
 import org.apache.flink.runtime.state.FunctionSnapshotContext;
 import org.apache.flink.runtime.state.StateInitializationContext;
 import org.apache.flink.runtime.state.StateSnapshotContextSynchronousImpl;
@@ -54,7 +55,6 @@
 import org.junit.runner.RunWith;
 import org.mockito.Matchers;
 import org.mockito.Mockito;
-import org.mockito.internal.util.reflection.Whitebox;
 import org.powermock.api.mockito.PowerMockito;
 import org.powermock.core.classloader.annotations.PrepareForTest;
 import org.powermock.modules.junit4.PowerMockRunner;

flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/FlinkKinesisProducerTest.java

@@ -48,6 +48,7 @@
 
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
+import static org.mockito.ArgumentMatchers.nullable;
 import static org.mockito.Matchers.any;
 import static org.mockito.Matchers.anyString;
 import static org.mockito.Mockito.doAnswer;
@@ -431,7 +432,7 @@ private static class DummyFlinkKinesisProducer<T> extends FlinkKinesisProducer<T
  // set up mock producer
  this.mockProducer = mock(KinesisProducer.class);
 
- when(mockProducer.addUserRecord(anyString(), anyString(), anyString(), any(ByteBuffer.class))).thenAnswer(new Answer<Object>() {
+ when(mockProducer.addUserRecord(anyString(), anyString(), nullable(String.class), any(ByteBuffer.class))).thenAnswer(new Answer<Object>() {
  @Override
  public Object answer(InvocationOnMock invocationOnMock) throws Throwable {
  SettableFuture<UserRecordResult> future = SettableFuture.create();

flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/proxy/KinesisProxyTest.java

@@ -63,9 +63,9 @@
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 import static org.mockito.Matchers.any;
-import static org.mockito.Matchers.argThat;
 import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.mock;
+import static org.mockito.hamcrest.MockitoHamcrest.argThat;
 
 /**
  * Test for methods in the {@link KinesisProxy} class.

flink-connectors/flink-hadoop-compatibility/src/test/java/org/apache/flink/api/java/hadoop/mapred/HadoopInputFormatTest.java

@@ -42,10 +42,10 @@
 import static org.hamcrest.CoreMatchers.equalTo;
 import static org.hamcrest.core.Is.is;
 import static org.junit.Assert.assertThat;
+import static org.mockito.ArgumentMatchers.nullable;
 import static org.mockito.Mockito.any;
 import static org.mockito.Mockito.anyInt;
 import static org.mockito.Mockito.anyLong;
-import static org.mockito.Mockito.anyString;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
@@ -151,7 +151,7 @@ public void testReachedEndWithNoElementsRemaining() throws IOException {
  @Test
  public void testFetchNext() throws IOException {
  DummyRecordReader recordReader = mock(DummyRecordReader.class);
- when(recordReader.next(anyString(), anyLong())).thenReturn(true);
+ when(recordReader.next(nullable(String.class), nullable(Long.class))).thenReturn(true);
 
  DummyInputFormat inputFormat = mock(DummyInputFormat.class);
  when(inputFormat.getRecordReader(any(InputSplit.class), any(JobConf.class), any(Reporter.class))).thenReturn(recordReader);
@@ -160,7 +160,7 @@ public void testFetchNext() throws IOException {
  hadoopInputFormat.open(getHadoopInputSplit());
  hadoopInputFormat.fetchNext();
 
- verify(recordReader, times(1)).next(anyString(), anyLong());
+ verify(recordReader, times(1)).next(nullable(String.class), anyLong());
  assertThat(hadoopInputFormat.hasNext, is(true));
  assertThat(hadoopInputFormat.fetched, is(true));
  }

flink-connectors/flink-hadoop-compatibility/src/test/java/org/apache/flink/api/java/hadoop/mapred/HadoopOutputFormatTest.java

@@ -38,6 +38,7 @@
 
 import java.io.IOException;
 
+import static org.mockito.ArgumentMatchers.nullable;
 import static org.mockito.Matchers.anyString;
 import static org.mockito.Mockito.any;
 import static org.mockito.Mockito.anyLong;
@@ -65,7 +66,7 @@ public void testOpen() throws Exception {
 
  verify(jobConf, times(2)).getOutputCommitter();
  verify(outputCommitter, times(1)).setupJob(any(JobContext.class));
- verify(dummyOutputFormat, times(1)).getRecordWriter(any(FileSystem.class), any(JobConf.class), anyString(), any(Progressable.class));
+ verify(dummyOutputFormat, times(1)).getRecordWriter(nullable(FileSystem.class), any(JobConf.class), anyString(), any(Progressable.class));
  }
 
  @Test
@@ -96,7 +97,7 @@ public void testConfigureWithJobConfigurable() {
  public void testCloseWithTaskCommit() throws Exception {
  OutputFormat<String, Long> dummyOutputFormat = mock(DummyOutputFormat.class);
  DummyOutputCommitter outputCommitter = mock(DummyOutputCommitter.class);
- when(outputCommitter.needsTaskCommit(any(TaskAttemptContext.class))).thenReturn(true);
+ when(outputCommitter.needsTaskCommit(nullable(TaskAttemptContext.class))).thenReturn(true);
  DummyRecordWriter recordWriter = mock(DummyRecordWriter.class);
  JobConf jobConf = mock(JobConf.class);
 
@@ -106,8 +107,8 @@ public void testCloseWithTaskCommit() throws Exception {
 
  outputFormat.close();
 
- verify(recordWriter, times(1)).close(any(Reporter.class));
- verify(outputCommitter, times(1)).commitTask(any(TaskAttemptContext.class));
+ verify(recordWriter, times(1)).close(nullable(Reporter.class));
+ verify(outputCommitter, times(1)).commitTask(nullable(TaskAttemptContext.class));
  }
 
  @Test

flink-connectors/flink-hadoop-compatibility/src/test/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopInputFormatTest.java

@@ -43,6 +43,7 @@
 import static org.hamcrest.CoreMatchers.equalTo;
 import static org.hamcrest.core.Is.is;
 import static org.junit.Assert.assertThat;
+import static org.mockito.ArgumentMatchers.nullable;
 import static org.mockito.Mockito.any;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
@@ -81,13 +82,13 @@ public void testCreateInputSplits() throws Exception {
  @Test
  public void testOpen() throws Exception {
  DummyInputFormat inputFormat = mock(DummyInputFormat.class);
- when(inputFormat.createRecordReader(any(InputSplit.class), any(TaskAttemptContext.class))).thenReturn(new DummyRecordReader());
+ when(inputFormat.createRecordReader(nullable(InputSplit.class), any(TaskAttemptContext.class))).thenReturn(new DummyRecordReader());
  HadoopInputSplit inputSplit = mock(HadoopInputSplit.class);
 
  HadoopInputFormat<String, Long> hadoopInputFormat = setupHadoopInputFormat(inputFormat, Job.getInstance(), null);
  hadoopInputFormat.open(inputSplit);
 
- verify(inputFormat, times(1)).createRecordReader(any(InputSplit.class), any(TaskAttemptContext.class));
+ verify(inputFormat, times(1)).createRecordReader(nullable(InputSplit.class), any(TaskAttemptContext.class));
  assertThat(hadoopInputFormat.fetched, is(false));
  }
 

flink-connectors/flink-hadoop-compatibility/src/test/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopOutputFormatTest.java

@@ -33,9 +33,8 @@
 
 import java.io.IOException;
 
+import static org.mockito.ArgumentMatchers.nullable;
 import static org.mockito.Matchers.any;
-import static org.mockito.Mockito.anyLong;
-import static org.mockito.Mockito.anyString;
 import static org.mockito.Mockito.doNothing;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
@@ -59,7 +58,7 @@ public void testWriteRecord() throws Exception {
 
  hadoopOutputFormat.writeRecord(new Tuple2<String, Long>());
 
- verify(recordWriter, times(1)).write(anyString(), anyLong());
+ verify(recordWriter, times(1)).write(nullable(String.class), nullable(Long.class));
  }
 
  @Test
@@ -89,8 +88,8 @@ public void testCloseWithNeedsTaskCommitTrue() throws Exception {
 
  hadoopOutputFormat.close();
 
- verify(outputCommitter, times(1)).commitTask(any(TaskAttemptContext.class));
- verify(recordWriter, times(1)).close(any(TaskAttemptContext.class));
+ verify(outputCommitter, times(1)).commitTask(nullable(TaskAttemptContext.class));
+ verify(recordWriter, times(1)).close(nullable(TaskAttemptContext.class));
  }
 
  @Test
@@ -104,8 +103,8 @@ public void testCloseWithNeedsTaskCommitFalse() throws Exception {
 
  hadoopOutputFormat.close();
 
- verify(outputCommitter, times(0)).commitTask(any(TaskAttemptContext.class));
- verify(recordWriter, times(1)).close(any(TaskAttemptContext.class));
+ verify(outputCommitter, times(0)).commitTask(nullable(TaskAttemptContext.class));
+ verify(recordWriter, times(1)).close(nullable(TaskAttemptContext.class));
  }
 
  @Test
@@ -134,7 +133,7 @@ public void testFinalizedGlobal() throws Exception {
 
  private OutputCommitter setupOutputCommitter(boolean needsTaskCommit) throws IOException {
  OutputCommitter outputCommitter = Mockito.mock(OutputCommitter.class);
- when(outputCommitter.needsTaskCommit(any(TaskAttemptContext.class))).thenReturn(needsTaskCommit);
+ when(outputCommitter.needsTaskCommit(nullable(TaskAttemptContext.class))).thenReturn(needsTaskCommit);
  doNothing().when(outputCommitter).commitTask(any(TaskAttemptContext.class));
 
  return outputCommitter;

flink-core/src/test/java/org/apache/flink/util/MethodForwardingTestUtil.java

@@ -98,8 +98,7 @@ public static <D, W, I extends D> void testMethodForwarding(
  I delegate = delegateObjectSupplier.get();
 
  //check if we need to wrap the delegate object as a spy, or if it is already testable with Mockito.
- MockUtil mockUtil = new MockUtil();
- if (!mockUtil.isSpy(delegate) || !mockUtil.isMock(delegate)) {
+ if (!MockUtil.isSpy(delegate) || !MockUtil.isMock(delegate)) {
  delegate = spy(delegate);
  }
 

flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroOutputFormatTest.java

@@ -24,6 +24,7 @@
 import org.apache.flink.formats.avro.generated.Colors;
 import org.apache.flink.formats.avro.generated.Fixed2;
 import org.apache.flink.formats.avro.generated.User;
+import org.apache.flink.mock.Whitebox;
 
 import org.apache.avro.Schema;
 import org.apache.avro.file.DataFileReader;
@@ -34,7 +35,6 @@
 import org.joda.time.LocalDate;
 import org.joda.time.LocalTime;
 import org.junit.Test;
-import org.mockito.internal.util.reflection.Whitebox;
 
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;

flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPOperatorTest.java

@@ -35,6 +35,7 @@
 import org.apache.flink.cep.pattern.conditions.IterativeCondition;
 import org.apache.flink.cep.pattern.conditions.SimpleCondition;
 import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
+import org.apache.flink.mock.Whitebox;
 import org.apache.flink.runtime.checkpoint.OperatorSubtaskState;
 import org.apache.flink.runtime.state.memory.MemoryStateBackend;
 import org.apache.flink.streaming.api.watermark.Watermark;
@@ -53,7 +54,6 @@
 import org.junit.Test;
 import org.junit.rules.TemporaryFolder;
 import org.mockito.Mockito;
-import org.mockito.internal.util.reflection.Whitebox;
 
 import java.util.ArrayList;
 import java.util.Collections;

flink-mesos/src/test/java/org/apache/flink/mesos/runtime/clusterframework/MesosResourceManagerTest.java

@@ -344,7 +344,7 @@ class MockResourceManagerRuntimeServices {
  doAnswer(new Answer<Object>() {
  @Override
  public Object answer(InvocationOnMock invocation) throws Throwable {
- rmActions = invocation.getArgumentAt(2, ResourceActions.class);
+ rmActions = invocation.getArgument(2);
  slotManagerStarted.complete(true);
  return null;
  }

flink-mesos/src/test/scala/org/apache/flink/mesos/Utils.scala

@@ -29,12 +29,9 @@ import scala.reflect.ClassTag
 
 object Matchers {
  def contentsMatch[T](plan: Seq[T]): java.util.Collection[T] = {
- org.mockito.Matchers.argThat(new ArgumentMatcher[java.util.Collection[T]] {
- override def matches(o: scala.Any): Boolean = o match {
- case actual: java.util.Collection[T] =>
+ org.mockito.ArgumentMatchers.argThat(new ArgumentMatcher[java.util.Collection[T]] {
+ def matches(actual: java.util.Collection[T]): Boolean =
  actual.size() == plan.size && actual.containsAll(plan.asJava)
- case _ => false
- }
  })
  }
 }

flink-mesos/src/test/scala/org/apache/flink/mesos/scheduler/LaunchCoordinatorTest.scala

@@ -37,7 +37,7 @@ import org.junit.runner.RunWith
 import org.mockito.Mockito.{verify, _}
 import org.mockito.invocation.InvocationOnMock
 import org.mockito.stubbing.Answer
-import org.mockito.{Matchers => MM, Mockito}
+import org.mockito.{ArgumentMatchers => MM, Mockito}
 import org.scalatest.junit.JUnitRunner
 import org.scalatest.{BeforeAndAfterAll, Matchers, WordSpecLike}