GJL closed pull request #7302: [FLINK-11156] [tests] Reconcile
ZooKeeperCompletedCheckpointStoreMockitoTest with JDK 9
URL: https://github.com/apache/flink/pull/7302
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStore.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStore.java
index 51f40085c3f..e87ca362331 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStore.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStore.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.checkpoint;
+import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.runtime.jobgraph.JobStatus;
import org.apache.flink.runtime.jobmanager.HighAvailabilityMode;
@@ -32,6 +33,7 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import javax.annotation.Nonnull;
import java.io.IOException;
import java.util.ArrayDeque;
import java.util.ArrayList;
@@ -107,34 +109,70 @@
* @throws Exception
*/
public ZooKeeperCompletedCheckpointStore(
- int maxNumberOfCheckpointsToRetain,
- CuratorFramework client,
- String checkpointsPath,
- RetrievableStateStorageHelper<CompletedCheckpoint>
stateStorage,
- Executor executor) throws Exception {
+ int maxNumberOfCheckpointsToRetain,
+ CuratorFramework client,
+ String checkpointsPath,
+ RetrievableStateStorageHelper<CompletedCheckpoint> stateStorage,
+ Executor executor
+ ) throws Exception {
+ this(maxNumberOfCheckpointsToRetain,
+ adaptNameSpace(client, checkpointsPath),
+ stateStorage,
+ executor);
- checkArgument(maxNumberOfCheckpointsToRetain >= 1, "Must retain
at least one checkpoint.");
- checkNotNull(stateStorage, "State storage");
-
- this.maxNumberOfCheckpointsToRetain =
maxNumberOfCheckpointsToRetain;
+ LOG.info("Initialized in '{}'.", checkpointsPath);
+ }
- checkNotNull(client, "Curator client");
- checkNotNull(checkpointsPath, "Checkpoints path");
+ @VisibleForTesting
+ ZooKeeperCompletedCheckpointStore(
+ int maxNumberOfCheckpointsToRetain,
+ CuratorFramework client,
+ String checkpointsPath,
+ Executor executor,
+ ZooKeeperStateHandleStore<CompletedCheckpoint>
checkpointsInZooKeeper
+ ) throws Exception {
+ this(maxNumberOfCheckpointsToRetain,
+ adaptNameSpace(client, checkpointsPath),
+ executor,
+ checkpointsInZooKeeper);
- // Ensure that the checkpoints path exists
- client.newNamespaceAwareEnsurePath(checkpointsPath)
- .ensure(client.getZookeeperClient());
+ LOG.info("Initialized in '{}'.", checkpointsPath);
+ }
- // All operations will have the path as root
- this.client = client.usingNamespace(client.getNamespace() +
checkpointsPath);
+ private ZooKeeperCompletedCheckpointStore(
+ int maxNumberOfCheckpointsToRetain,
+ CuratorFramework client,
+ RetrievableStateStorageHelper<CompletedCheckpoint> stateStorage,
+ Executor executor
+ ) {
+ this(maxNumberOfCheckpointsToRetain,
+ client,
+ executor,
+ new ZooKeeperStateHandleStore<>(client, stateStorage));
+ }
- this.checkpointsInZooKeeper = new
ZooKeeperStateHandleStore<>(this.client, stateStorage);
+ private ZooKeeperCompletedCheckpointStore(
+ int maxNumberOfCheckpointsToRetain,
+ @Nonnull CuratorFramework client,
+ @Nonnull Executor executor,
+ @Nonnull ZooKeeperStateHandleStore<CompletedCheckpoint>
checkpointsInZooKeeper
+ ) {
+ checkArgument(maxNumberOfCheckpointsToRetain >= 1, "Must retain
at least one checkpoint.");
+ this.maxNumberOfCheckpointsToRetain =
maxNumberOfCheckpointsToRetain;
+ this.client = client;
+ this.executor = executor;
+ this.checkpointsInZooKeeper = checkpointsInZooKeeper;
this.completedCheckpoints = new
ArrayDeque<>(maxNumberOfCheckpointsToRetain + 1);
+ }
- this.executor = checkNotNull(executor);
+ private static CuratorFramework adaptNameSpace(CuratorFramework client,
String checkpointsPath) throws Exception {
+ // Ensure that the checkpoints path exists
+ client.newNamespaceAwareEnsurePath(checkpointsPath)
+ .ensure(client.getZookeeperClient());
- LOG.info("Initialized in '{}'.", checkpointsPath);
+ // All operations will have the path as root
+ return client.usingNamespace(client.getNamespace() +
checkpointsPath);
}
@Override
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStoreMockitoTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStoreMockitoTest.java
index e9b90b72611..5fe0f41fc52 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStoreMockitoTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStoreMockitoTest.java
@@ -34,12 +34,9 @@
import org.apache.curator.framework.api.ErrorListenerPathable;
import org.apache.curator.utils.EnsurePath;
import org.junit.Test;
-import org.junit.runner.RunWith;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
-import org.powermock.core.classloader.annotations.PrepareForTest;
-import org.powermock.modules.junit4.PowerMockRunner;
import java.io.IOException;
import java.util.ArrayList;
@@ -55,21 +52,18 @@
import static org.junit.Assert.assertTrue;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.anyString;
+import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
-import static org.powermock.api.mockito.PowerMockito.doAnswer;
-import static org.powermock.api.mockito.PowerMockito.doThrow;
-import static org.powermock.api.mockito.PowerMockito.whenNew;
/**
* Mockito based tests for the {@link ZooKeeperStateHandleStore}.
*/
-@RunWith(PowerMockRunner.class)
-@PrepareForTest(ZooKeeperCompletedCheckpointStore.class)
public class ZooKeeperCompletedCheckpointStoreMockitoTest extends TestLogger {
/**
@@ -125,7 +119,6 @@ public void testCheckpointRecovery() throws Exception {
final RetrievableStateStorageHelper<CompletedCheckpoint>
storageHelperMock = mock(RetrievableStateStorageHelper.class);
ZooKeeperStateHandleStore<CompletedCheckpoint>
zooKeeperStateHandleStoreMock = spy(new ZooKeeperStateHandleStore<>(client,
storageHelperMock));
-
whenNew(ZooKeeperStateHandleStore.class).withAnyArguments().thenReturn(zooKeeperStateHandleStoreMock);
doReturn(checkpointsInZooKeeper).when(zooKeeperStateHandleStoreMock).getAllAndLock();
final int numCheckpointsToRetain = 1;
@@ -165,14 +158,13 @@ public Void answer(InvocationOnMock invocation) throws
Throwable {
});
final String checkpointsPath = "foobar";
- final RetrievableStateStorageHelper<CompletedCheckpoint>
stateStorage = mock(RetrievableStateStorageHelper.class);
ZooKeeperCompletedCheckpointStore
zooKeeperCompletedCheckpointStore = new ZooKeeperCompletedCheckpointStore(
numCheckpointsToRetain,
client,
checkpointsPath,
- stateStorage,
- Executors.directExecutor());
+ Executors.directExecutor(),
+ zooKeeperStateHandleStoreMock);
zooKeeperCompletedCheckpointStore.recover();
@@ -213,7 +205,6 @@ public void testAddCheckpointWithFailedRemove() throws
Exception {
ZooKeeperStateHandleStore<CompletedCheckpoint>
zookeeperStateHandleStoreMock =
spy(new ZooKeeperStateHandleStore<>(client,
storageHelperMock));
-
whenNew(ZooKeeperStateHandleStore.class).withAnyArguments().thenReturn(zookeeperStateHandleStoreMock);
doAnswer(new
Answer<RetrievableStateHandle<CompletedCheckpoint>>() {
@Override
@@ -231,14 +222,13 @@ public void testAddCheckpointWithFailedRemove() throws
Exception {
final int numCheckpointsToRetain = 1;
final String checkpointsPath = "foobar";
- final RetrievableStateStorageHelper<CompletedCheckpoint>
stateSotrage = mock(RetrievableStateStorageHelper.class);
ZooKeeperCompletedCheckpointStore
zooKeeperCompletedCheckpointStore = new ZooKeeperCompletedCheckpointStore(
numCheckpointsToRetain,
client,
checkpointsPath,
- stateSotrage,
- Executors.directExecutor());
+ Executors.directExecutor(),
+ zookeeperStateHandleStoreMock);
for (long i = 0; i <= numCheckpointsToRetain; ++i) {
CompletedCheckpoint checkpointToAdd =
mock(CompletedCheckpoint.class);
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
