You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tr...@apache.org on 2017/10/05 14:06:56 UTC
[03/14] flink git commit: [FLINK-7068][blob] change BlobService
sub-classes for permanent and transient BLOBs
http://git-wip-us.apache.org/repos/asf/flink/blob/071e27f7/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/InterruptSensitiveRestoreTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/InterruptSensitiveRestoreTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/InterruptSensitiveRestoreTest.java
index f73499c..1095316 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/InterruptSensitiveRestoreTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/InterruptSensitiveRestoreTest.java
@@ -26,6 +26,8 @@ import org.apache.flink.core.fs.FSDataInputStream;
import org.apache.flink.core.testutils.OneShotLatch;
import org.apache.flink.runtime.blob.BlobCache;
import org.apache.flink.runtime.blob.BlobKey;
+import org.apache.flink.runtime.blob.PermanentBlobCache;
+import org.apache.flink.runtime.blob.TransientBlobCache;
import org.apache.flink.runtime.broadcast.BroadcastVariableManager;
import org.apache.flink.runtime.checkpoint.OperatorSubtaskState;
import org.apache.flink.runtime.checkpoint.TaskStateSnapshot;
@@ -240,6 +242,13 @@ public class InterruptSensitiveRestoreTest {
SourceStreamTask.class.getName(),
taskConfig);
+ BlobCache blobCache = mock(BlobCache.class);
+ PermanentBlobCache permanentBlobCache = mock(PermanentBlobCache.class);
+ TransientBlobCache transientBlobCache = mock(TransientBlobCache.class);
+
+ when(blobCache.getPermanentBlobStore()).thenReturn(permanentBlobCache);
+ when(blobCache.getTransientBlobStore()).thenReturn(transientBlobCache);
+
return new Task(
jobInformation,
taskInformation,
@@ -258,7 +267,7 @@ public class InterruptSensitiveRestoreTest {
mock(TaskManagerActions.class),
mock(InputSplitProvider.class),
mock(CheckpointResponder.class),
- mock(BlobCache.class),
+ blobCache,
new FallbackLibraryCacheManager(),
new FileCache(new String[] { EnvironmentInformation.getTemporaryFileDirectory() }),
new TestingTaskManagerRuntimeInfo(),
http://git-wip-us.apache.org/repos/asf/flink/blob/071e27f7/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTerminationTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTerminationTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTerminationTest.java
index 79e9583..9e0468b 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTerminationTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTerminationTest.java
@@ -25,6 +25,8 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.testutils.OneShotLatch;
import org.apache.flink.runtime.blob.BlobCache;
import org.apache.flink.runtime.blob.BlobKey;
+import org.apache.flink.runtime.blob.PermanentBlobCache;
+import org.apache.flink.runtime.blob.TransientBlobCache;
import org.apache.flink.runtime.broadcast.BroadcastVariableManager;
import org.apache.flink.runtime.checkpoint.CheckpointOptions;
import org.apache.flink.runtime.clusterframework.types.AllocationID;
@@ -136,6 +138,13 @@ public class StreamTaskTerminationTest extends TestLogger {
final NetworkEnvironment networkEnv = mock(NetworkEnvironment.class);
when(networkEnv.createKvStateTaskRegistry(any(JobID.class), any(JobVertexID.class))).thenReturn(mock(TaskKvStateRegistry.class));
+ BlobCache blobCache = mock(BlobCache.class);
+ PermanentBlobCache permanentBlobCache = mock(PermanentBlobCache.class);
+ TransientBlobCache transientBlobCache = mock(TransientBlobCache.class);
+
+ when(blobCache.getPermanentBlobStore()).thenReturn(permanentBlobCache);
+ when(blobCache.getTransientBlobStore()).thenReturn(transientBlobCache);
+
final Task task = new Task(
jobInformation,
taskInformation,
@@ -154,7 +163,7 @@ public class StreamTaskTerminationTest extends TestLogger {
mock(TaskManagerActions.class),
mock(InputSplitProvider.class),
mock(CheckpointResponder.class),
- mock(BlobCache.class),
+ blobCache,
new FallbackLibraryCacheManager(),
mock(FileCache.class),
taskManagerRuntimeInfo,
http://git-wip-us.apache.org/repos/asf/flink/blob/071e27f7/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java
index 811d700..100c16b 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java
@@ -29,6 +29,8 @@ import org.apache.flink.core.fs.CloseableRegistry;
import org.apache.flink.core.testutils.OneShotLatch;
import org.apache.flink.runtime.blob.BlobCache;
import org.apache.flink.runtime.blob.BlobKey;
+import org.apache.flink.runtime.blob.PermanentBlobCache;
+import org.apache.flink.runtime.blob.TransientBlobCache;
import org.apache.flink.runtime.broadcast.BroadcastVariableManager;
import org.apache.flink.runtime.checkpoint.CheckpointMetaData;
import org.apache.flink.runtime.checkpoint.CheckpointMetrics;
@@ -859,6 +861,12 @@ public class StreamTaskTest extends TestLogger {
Configuration taskManagerConfig) throws Exception {
BlobCache blobCache = mock(BlobCache.class);
+ PermanentBlobCache permanentBlobCache = mock(PermanentBlobCache.class);
+ TransientBlobCache transientBlobCache = mock(TransientBlobCache.class);
+
+ when(blobCache.getPermanentBlobStore()).thenReturn(permanentBlobCache);
+ when(blobCache.getTransientBlobStore()).thenReturn(transientBlobCache);
+
LibraryCacheManager libCache = mock(LibraryCacheManager.class);
when(libCache.getClassLoader(any(JobID.class))).thenReturn(StreamTaskTest.class.getClassLoader());