You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2016/05/26 19:17:25 UTC
[2/3] flink git commit: [hotfix] Fix access to temp file directories
in SpillingAdaptiveSpanningRecordDeserializer
[hotfix] Fix access to temp file directories in SpillingAdaptiveSpanningRecordDeserializer
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/bf256c7f
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/bf256c7f
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/bf256c7f
Branch: refs/heads/master
Commit: bf256c7fbe05accdadc8470013879f567341d1aa
Parents: 5a7f4e3
Author: Stephan Ewen <se...@apache.org>
Authored: Wed May 25 17:04:31 2016 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Thu May 26 21:15:20 2016 +0200
----------------------------------------------------------------------
.../api/common/functions/RuntimeContext.java | 2 +-
.../org/apache/flink/metrics/MetricGroup.java | 6 +--
.../clusterframework/types/ResourceID.java | 34 +++++++--------
.../runtime/io/disk/iomanager/IOManager.java | 15 ++++++-
.../api/reader/AbstractRecordReader.java | 12 +++++-
.../network/api/reader/MutableRecordReader.java | 12 +++++-
.../io/network/api/reader/RecordReader.java | 17 +++++---
...llingAdaptiveSpanningRecordDeserializer.java | 16 ++-----
.../task/IterationSynchronizationSinkTask.java | 6 ++-
.../flink/runtime/operators/BatchTask.java | 18 +++++---
.../flink/runtime/operators/DataSinkTask.java | 8 +++-
.../taskmanager/TaskManagerRuntimeInfo.java | 35 ++++++++++++++--
.../flink/runtime/taskmanager/TaskManager.scala | 3 +-
.../SpanningRecordSerializationTest.java | 4 +-
.../network/serialization/LargeRecordsTest.java | 5 ++-
.../SlotCountExceedingParallelismTest.java | 5 ++-
.../operators/drivers/TestTaskContext.java | 3 +-
.../testutils/BinaryOperatorTestBase.java | 3 +-
.../operators/testutils/DriverTestBase.java | 3 +-
.../operators/testutils/MockEnvironment.java | 5 ++-
.../testutils/UnaryOperatorTestBase.java | 3 +-
.../runtime/taskmanager/TaskAsyncCallTest.java | 2 +-
.../runtime/taskmanager/TaskCancelTest.java | 10 +++--
.../flink/runtime/taskmanager/TaskTest.java | 2 +-
.../apache/flink/runtime/jobmanager/Tasks.scala | 44 ++++++++++++++++----
.../runtime/io/StreamInputProcessor.java | 3 +-
.../runtime/io/StreamTwoInputProcessor.java | 3 +-
.../runtime/tasks/StreamMockEnvironment.java | 4 +-
.../streaming/runtime/tasks/StreamTaskTest.java | 2 +-
.../runtime/NetworkStackThroughputITCase.java | 20 ++++++---
30 files changed, 212 insertions(+), 93 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/bf256c7f/flink-core/src/main/java/org/apache/flink/api/common/functions/RuntimeContext.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/functions/RuntimeContext.java b/flink-core/src/main/java/org/apache/flink/api/common/functions/RuntimeContext.java
index ed2f613..9a04b24 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/functions/RuntimeContext.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/functions/RuntimeContext.java
@@ -63,7 +63,7 @@ public interface RuntimeContext {
* Returns the metric group for this parallel subtask.
*
* @return The metric group for this parallel subtask.
- */
+ */
MetricGroup getMetricGroup();
/**
http://git-wip-us.apache.org/repos/asf/flink/blob/bf256c7f/flink-core/src/main/java/org/apache/flink/metrics/MetricGroup.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/metrics/MetricGroup.java b/flink-core/src/main/java/org/apache/flink/metrics/MetricGroup.java
index a3832ff..6c9e044 100644
--- a/flink-core/src/main/java/org/apache/flink/metrics/MetricGroup.java
+++ b/flink-core/src/main/java/org/apache/flink/metrics/MetricGroup.java
@@ -21,14 +21,14 @@ package org.apache.flink.metrics;
import org.apache.flink.annotation.PublicEvolving;
/**
- * A MetricGroup is a named container for {@link Metric Metrics} and {@link MetricGroup MetricGroups}.
+ * A MetricGroup is a named container for {@link Metric Metrics} and further metric subgroups.
*
* <p>Instances of this class can be used to register new metrics with Flink and to create a nested
* hierarchy based on the group names.
*
* <p>A MetricGroup is uniquely identified by it's place in the hierarchy and name.
*
- * <p>Metrics groups can be {@link #close() closed}. Upon closing, they de-register all metrics
+ * <p>Metrics groups can be {@link #close() closed}. Upon closing, the group de-register all metrics
* from any metrics reporter and any internal maps. Note that even closed metrics groups
* return Counters, Gauges, etc to the code, to prevent exceptions in the monitored code.
* These metrics simply do not get reported any more, when created on a closed group.
@@ -39,7 +39,7 @@ public interface MetricGroup {
// ------------------------------------------------------------------------
// Closing
// ------------------------------------------------------------------------
-
+
/**
* Marks the group as closed.
* Recursively unregisters all {@link Metric Metrics} contained in this group.
http://git-wip-us.apache.org/repos/asf/flink/blob/bf256c7f/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/types/ResourceID.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/types/ResourceID.java b/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/types/ResourceID.java
index e599456..9d82c76 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/types/ResourceID.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/types/ResourceID.java
@@ -45,23 +45,11 @@ public class ResourceID implements Serializable {
return resourceId;
}
- /**
- * Generate a random resource id.
- * @return A random resource id.
- */
- public static ResourceID generate() {
- return new ResourceID(new AbstractID().toString());
- }
-
@Override
public final boolean equals(Object o) {
- if (this == o) {
- return true;
- } else if (o == null || !(o instanceof ResourceID)) {
- return false;
- } else {
- return resourceId.equals(((ResourceID) o).resourceId);
- }
+ return this == o ||
+ (o != null && o.getClass() == ResourceID.class &&
+ this.resourceId.equals(((ResourceID) o).resourceId));
}
@Override
@@ -71,8 +59,18 @@ public class ResourceID implements Serializable {
@Override
public String toString() {
- return "ResourceID{" +
- "resourceId='" + resourceId + '\'' +
- '}';
+ return "ResourceID (" + resourceId + ')';
+ }
+
+ // ------------------------------------------------------------------------
+ // factory
+ // ------------------------------------------------------------------------
+
+ /**
+ * Generate a random resource id.
+ * @return A random resource id.
+ */
+ public static ResourceID generate() {
+ return new ResourceID(new AbstractID().toString());
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/bf256c7f/flink-runtime/src/main/java/org/apache/flink/runtime/io/disk/iomanager/IOManager.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/disk/iomanager/IOManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/disk/iomanager/IOManager.java
index 0942f72..7904cc4 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/disk/iomanager/IOManager.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/disk/iomanager/IOManager.java
@@ -21,6 +21,7 @@ package org.apache.flink.runtime.io.disk.iomanager;
import org.apache.commons.io.FileUtils;
import org.apache.flink.core.memory.MemorySegment;
import org.apache.flink.runtime.io.network.buffer.Buffer;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -281,7 +282,19 @@ public abstract class IOManager {
public File[] getSpillingDirectories() {
return this.paths;
}
-
+
+ /**
+ * Gets the directories that the I/O manager spills to, as path strings.
+ *
+ * @return The directories that the I/O manager spills to, as path strings.
+ */
+ public String[] getSpillingDirectoriesPaths() {
+ String[] strings = new String[this.paths.length];
+ for (int i = 0; i < strings.length; i++) {
+ strings[i] = paths[i].getAbsolutePath();
+ }
+ return strings;
+ }
protected int getNextPathNum() {
final int next = this.nextPath;
http://git-wip-us.apache.org/repos/asf/flink/blob/bf256c7f/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/reader/AbstractRecordReader.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/reader/AbstractRecordReader.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/reader/AbstractRecordReader.java
index a784f54..48ac558 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/reader/AbstractRecordReader.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/reader/AbstractRecordReader.java
@@ -45,14 +45,22 @@ abstract class AbstractRecordReader<T extends IOReadableWritable> extends Abstra
private boolean isFinished;
+ /**
+ * Creates a new AbstractRecordReader that de-serializes records from the given input gate and
+ * can spill partial records to disk, if they grow large.
+ *
+ * @param inputGate The input gate to read from.
+ * @param tmpDirectories The temp directories. USed for spilling if the reader concurrently
+ * reconstructs multiple large records.
+ */
@SuppressWarnings("unchecked")
- protected AbstractRecordReader(InputGate inputGate) {
+ protected AbstractRecordReader(InputGate inputGate, String[] tmpDirectories) {
super(inputGate);
// Initialize one deserializer per input channel
this.recordDeserializers = new SpillingAdaptiveSpanningRecordDeserializer[inputGate.getNumberOfInputChannels()];
for (int i = 0; i < recordDeserializers.length; i++) {
- recordDeserializers[i] = new SpillingAdaptiveSpanningRecordDeserializer<T>();
+ recordDeserializers[i] = new SpillingAdaptiveSpanningRecordDeserializer<T>(tmpDirectories);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/bf256c7f/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/reader/MutableRecordReader.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/reader/MutableRecordReader.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/reader/MutableRecordReader.java
index d7cc7e9..9836ba4 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/reader/MutableRecordReader.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/reader/MutableRecordReader.java
@@ -25,8 +25,16 @@ import java.io.IOException;
public class MutableRecordReader<T extends IOReadableWritable> extends AbstractRecordReader<T> implements MutableReader<T> {
- public MutableRecordReader(InputGate inputGate) {
- super(inputGate);
+ /**
+ * Creates a new MutableRecordReader that de-serializes records from the given input gate and
+ * can spill partial records to disk, if they grow large.
+ *
+ * @param inputGate The input gate to read from.
+ * @param tmpDirectories The temp directories. USed for spilling if the reader concurrently
+ * reconstructs multiple large records.
+ */
+ public MutableRecordReader(InputGate inputGate, String[] tmpDirectories) {
+ super(inputGate, tmpDirectories);
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/bf256c7f/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/reader/RecordReader.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/reader/RecordReader.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/reader/RecordReader.java
index d45920e..9eed374 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/reader/RecordReader.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/reader/RecordReader.java
@@ -29,8 +29,16 @@ public class RecordReader<T extends IOReadableWritable> extends AbstractRecordRe
private T currentRecord;
- public RecordReader(InputGate inputGate, Class<T> recordType) {
- super(inputGate);
+ /**
+ * Creates a new RecordReader that de-serializes records from the given input gate and
+ * can spill partial records to disk, if they grow large.
+ *
+ * @param inputGate The input gate to read from.
+ * @param tmpDirectories The temp directories. USed for spilling if the reader concurrently
+ * reconstructs multiple large records.
+ */
+ public RecordReader(InputGate inputGate, Class<T> recordType, String[] tmpDirectories) {
+ super(inputGate, tmpDirectories);
this.recordType = recordType;
}
@@ -73,10 +81,7 @@ public class RecordReader<T extends IOReadableWritable> extends AbstractRecordRe
try {
return recordType.newInstance();
}
- catch (InstantiationException e) {
- throw new RuntimeException("Cannot instantiate class " + recordType.getName(), e);
- }
- catch (IllegalAccessException e) {
+ catch (InstantiationException | IllegalAccessException e) {
throw new RuntimeException("Cannot instantiate class " + recordType.getName(), e);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/bf256c7f/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/SpillingAdaptiveSpanningRecordDeserializer.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/SpillingAdaptiveSpanningRecordDeserializer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/SpillingAdaptiveSpanningRecordDeserializer.java
index 49f7584..7e96390 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/SpillingAdaptiveSpanningRecordDeserializer.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/SpillingAdaptiveSpanningRecordDeserializer.java
@@ -18,8 +18,6 @@
package org.apache.flink.runtime.io.network.api.serialization;
-import org.apache.flink.configuration.ConfigConstants;
-import org.apache.flink.configuration.GlobalConfiguration;
import org.apache.flink.core.io.IOReadableWritable;
import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.core.memory.DataInputViewStreamWrapper;
@@ -65,18 +63,12 @@ public class SpillingAdaptiveSpanningRecordDeserializer<T extends IOReadableWrit
private AccumulatorRegistry.Reporter reporter;
- private transient Counter numRecordsIn;
- private transient Counter numBytesIn;
+ private Counter numRecordsIn;
+ private Counter numBytesIn;
- public SpillingAdaptiveSpanningRecordDeserializer() {
-
- String tempDirString = GlobalConfiguration.getString(
- ConfigConstants.TASK_MANAGER_TMP_DIR_KEY,
- ConfigConstants.DEFAULT_TASK_MANAGER_TMP_PATH);
- String[] directories = tempDirString.split(",|" + File.pathSeparator);
-
+ public SpillingAdaptiveSpanningRecordDeserializer(String[] tmpDirectories) {
this.nonSpanningWrapper = new NonSpanningWrapper();
- this.spanningWrapper = new SpanningWrapper(directories);
+ this.spanningWrapper = new SpanningWrapper(tmpDirectories);
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/bf256c7f/flink-runtime/src/main/java/org/apache/flink/runtime/iterative/task/IterationSynchronizationSinkTask.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/iterative/task/IterationSynchronizationSinkTask.java b/flink-runtime/src/main/java/org/apache/flink/runtime/iterative/task/IterationSynchronizationSinkTask.java
index 2b710d2..f1ab93f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/iterative/task/IterationSynchronizationSinkTask.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/iterative/task/IterationSynchronizationSinkTask.java
@@ -72,7 +72,9 @@ public class IterationSynchronizationSinkTask extends AbstractInvokable implemen
@Override
public void invoke() throws Exception {
- this.headEventReader = new MutableRecordReader<IntValue>(getEnvironment().getInputGate(0));
+ this.headEventReader = new MutableRecordReader<IntValue>(
+ getEnvironment().getInputGate(0),
+ getEnvironment().getTaskManagerInfo().getTmpDirectories());
TaskConfig taskConfig = new TaskConfig(getTaskConfiguration());
@@ -184,7 +186,7 @@ public class IterationSynchronizationSinkTask extends AbstractInvokable implemen
// read (and thereby process all events in the handler's event handling functions)
try {
- while (this.headEventReader.next(rec)) {
+ if (this.headEventReader.next(rec)) {
throw new RuntimeException("Synchronization task must not see any records!");
}
} catch (InterruptedException iex) {
http://git-wip-us.apache.org/repos/asf/flink/blob/bf256c7f/flink-runtime/src/main/java/org/apache/flink/runtime/operators/BatchTask.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/BatchTask.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/BatchTask.java
index 546193c..f38b988 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/BatchTask.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/BatchTask.java
@@ -659,14 +659,18 @@ public class BatchTask<S extends Function, OT> extends AbstractInvokable impleme
if (groupSize == 1) {
// non-union case
- inputReaders[i] = new MutableRecordReader<IOReadableWritable>(getEnvironment().getInputGate(currentReaderOffset));
+ inputReaders[i] = new MutableRecordReader<IOReadableWritable>(
+ getEnvironment().getInputGate(currentReaderOffset),
+ getEnvironment().getTaskManagerInfo().getTmpDirectories());
} else if (groupSize > 1){
// union case
InputGate[] readers = new InputGate[groupSize];
for (int j = 0; j < groupSize; ++j) {
readers[j] = getEnvironment().getInputGate(currentReaderOffset + j);
}
- inputReaders[i] = new MutableRecordReader<IOReadableWritable>(new UnionInputGate(readers));
+ inputReaders[i] = new MutableRecordReader<IOReadableWritable>(
+ new UnionInputGate(readers),
+ getEnvironment().getTaskManagerInfo().getTmpDirectories());
} else {
throw new Exception("Illegal input group size in task configuration: " + groupSize);
}
@@ -701,14 +705,18 @@ public class BatchTask<S extends Function, OT> extends AbstractInvokable impleme
final int groupSize = this.config.getBroadcastGroupSize(i);
if (groupSize == 1) {
// non-union case
- broadcastInputReaders[i] = new MutableRecordReader<IOReadableWritable>(getEnvironment().getInputGate(currentReaderOffset));
+ broadcastInputReaders[i] = new MutableRecordReader<IOReadableWritable>(
+ getEnvironment().getInputGate(currentReaderOffset),
+ getEnvironment().getTaskManagerInfo().getTmpDirectories());
} else if (groupSize > 1){
// union case
InputGate[] readers = new InputGate[groupSize];
for (int j = 0; j < groupSize; ++j) {
readers[j] = getEnvironment().getInputGate(currentReaderOffset + j);
}
- broadcastInputReaders[i] = new MutableRecordReader<IOReadableWritable>(new UnionInputGate(readers));
+ broadcastInputReaders[i] = new MutableRecordReader<IOReadableWritable>(
+ new UnionInputGate(readers),
+ getEnvironment().getTaskManagerInfo().getTmpDirectories());
} else {
throw new Exception("Illegal input group size in task configuration: " + groupSize);
}
@@ -765,8 +773,6 @@ public class BatchTask<S extends Function, OT> extends AbstractInvokable impleme
*
* NOTE: This method must be invoked after the invocation of {@code #initInputReaders()} and
* {@code #initInputSerializersAndComparators(int)}!
- *
- * @param numInputs
*/
protected void initLocalStrategies(int numInputs) throws Exception {
http://git-wip-us.apache.org/repos/asf/flink/blob/bf256c7f/flink-runtime/src/main/java/org/apache/flink/runtime/operators/DataSinkTask.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/DataSinkTask.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/DataSinkTask.java
index 39bf23f..380edd4 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/DataSinkTask.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/DataSinkTask.java
@@ -332,10 +332,14 @@ public class DataSinkTask<IT> extends AbstractInvokable {
numGates += groupSize;
if (groupSize == 1) {
// non-union case
- inputReader = new MutableRecordReader<DeserializationDelegate<IT>>(getEnvironment().getInputGate(0));
+ inputReader = new MutableRecordReader<DeserializationDelegate<IT>>(
+ getEnvironment().getInputGate(0),
+ getEnvironment().getTaskManagerInfo().getTmpDirectories());
} else if (groupSize > 1){
// union case
- inputReader = new MutableRecordReader<IOReadableWritable>(new UnionInputGate(getEnvironment().getAllInputGates()));
+ inputReader = new MutableRecordReader<IOReadableWritable>(
+ new UnionInputGate(getEnvironment().getAllInputGates()),
+ getEnvironment().getTaskManagerInfo().getTmpDirectories());
} else {
throw new Exception("Illegal input group size in task configuration: " + groupSize);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/bf256c7f/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/TaskManagerRuntimeInfo.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/TaskManagerRuntimeInfo.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/TaskManagerRuntimeInfo.java
index 8d06f10..9ac982e 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/TaskManagerRuntimeInfo.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/TaskManagerRuntimeInfo.java
@@ -20,6 +20,9 @@ package org.apache.flink.runtime.taskmanager;
import org.apache.flink.configuration.Configuration;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkArgument;
+
/**
* Encapsulation of TaskManager runtime information, like hostname and configuration.
*/
@@ -33,14 +36,32 @@ public class TaskManagerRuntimeInfo implements java.io.Serializable {
/** configuration that the TaskManager was started with */
private final Configuration configuration;
+ /** list of temporary file directories */
+ private final String[] tmpDirectories;
+
/**
* Creates a runtime info.
+ *
* @param hostname The host name of the interface that the TaskManager uses to communicate.
* @param configuration The configuration that the TaskManager was started with.
+ * @param tmpDirectory The temporary file directory.
*/
- public TaskManagerRuntimeInfo(String hostname, Configuration configuration) {
- this.hostname = hostname;
- this.configuration = configuration;
+ public TaskManagerRuntimeInfo(String hostname, Configuration configuration, String tmpDirectory) {
+ this(hostname, configuration, new String[] { tmpDirectory });
+ }
+
+ /**
+ * Creates a runtime info.
+ * @param hostname The host name of the interface that the TaskManager uses to communicate.
+ * @param configuration The configuration that the TaskManager was started with.
+ * @param tmpDirectories The list of temporary file directories.
+ */
+ public TaskManagerRuntimeInfo(String hostname, Configuration configuration, String[] tmpDirectories) {
+ checkArgument(tmpDirectories.length > 0);
+ this.hostname = checkNotNull(hostname);
+ this.configuration = checkNotNull(configuration);
+ this.tmpDirectories = tmpDirectories;
+
}
/**
@@ -58,4 +79,12 @@ public class TaskManagerRuntimeInfo implements java.io.Serializable {
public Configuration getConfiguration() {
return configuration;
}
+
+ /**
+ * Gets the list of temporary file directories.
+ * @return The list of temporary file directories.
+ */
+ public String[] getTmpDirectories() {
+ return tmpDirectories;
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/bf256c7f/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
index a5cc18d..eb7a0ef 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
@@ -187,7 +187,8 @@ class TaskManager(
private val runtimeInfo = new TaskManagerRuntimeInfo(
connectionInfo.getHostname(),
- new UnmodifiableConfiguration(config.configuration))
+ new UnmodifiableConfiguration(config.configuration),
+ config.tmpDirPaths)
// --------------------------------------------------------------------------
// Actor messages and life cycle
// --------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/bf256c7f/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/api/serialization/SpanningRecordSerializationTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/api/serialization/SpanningRecordSerializationTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/api/serialization/SpanningRecordSerializationTest.java
index 819a94f..9d0ee67 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/api/serialization/SpanningRecordSerializationTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/api/serialization/SpanningRecordSerializationTest.java
@@ -106,7 +106,9 @@ public class SpanningRecordSerializationTest {
private void testSpillingDeserializer(Util.MockRecords records, int segmentSize) throws Exception {
RecordSerializer<SerializationTestType> serializer = new SpanningRecordSerializer<SerializationTestType>();
- RecordDeserializer<SerializationTestType> deserializer = new SpillingAdaptiveSpanningRecordDeserializer<SerializationTestType>();
+ RecordDeserializer<SerializationTestType> deserializer =
+ new SpillingAdaptiveSpanningRecordDeserializer<SerializationTestType>(
+ new String[] { System.getProperty("java.io.tmpdir") });
test(records, segmentSize, serializer, deserializer);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/bf256c7f/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/serialization/LargeRecordsTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/serialization/LargeRecordsTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/serialization/LargeRecordsTest.java
index d628596..1574fe9 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/serialization/LargeRecordsTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/serialization/LargeRecordsTest.java
@@ -147,7 +147,10 @@ public class LargeRecordsTest {
final int SEGMENT_SIZE = 32 * 1024;
final RecordSerializer<SerializationTestType> serializer = new SpanningRecordSerializer<SerializationTestType>();
- final RecordDeserializer<SerializationTestType> deserializer = new SpillingAdaptiveSpanningRecordDeserializer<SerializationTestType>();
+
+ final RecordDeserializer<SerializationTestType> deserializer =
+ new SpillingAdaptiveSpanningRecordDeserializer<SerializationTestType>(
+ new String[] { System.getProperty("java.io.tmpdir") } );
final Buffer buffer = new Buffer(MemorySegmentFactory.allocateUnpooledSegment(SEGMENT_SIZE), mock(BufferRecycler.class));
http://git-wip-us.apache.org/repos/asf/flink/blob/bf256c7f/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/SlotCountExceedingParallelismTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/SlotCountExceedingParallelismTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/SlotCountExceedingParallelismTest.java
index 561bda3..e12faf9 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/SlotCountExceedingParallelismTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/SlotCountExceedingParallelismTest.java
@@ -18,7 +18,6 @@
package org.apache.flink.runtime.jobmanager;
-import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.runtime.client.JobExecutionException;
import org.apache.flink.runtime.io.network.api.reader.RecordReader;
import org.apache.flink.runtime.io.network.api.writer.RecordWriter;
@@ -30,6 +29,7 @@ import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
import org.apache.flink.runtime.testingUtils.TestingCluster;
import org.apache.flink.runtime.testingUtils.TestingUtils;
import org.apache.flink.types.IntValue;
+
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
@@ -152,7 +152,8 @@ public class SlotCountExceedingParallelismTest {
public void invoke() throws Exception {
RecordReader<IntValue> reader = new RecordReader<>(
getEnvironment().getInputGate(0),
- IntValue.class);
+ IntValue.class,
+ getEnvironment().getTaskManagerInfo().getTmpDirectories());
try {
final int numberOfSubtaskIndexesToReceive = getTaskConfiguration().getInteger(CONFIG_KEY, 0);
http://git-wip-us.apache.org/repos/asf/flink/blob/bf256c7f/flink-runtime/src/test/java/org/apache/flink/runtime/operators/drivers/TestTaskContext.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/drivers/TestTaskContext.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/drivers/TestTaskContext.java
index 0300a07..15ad353 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/drivers/TestTaskContext.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/drivers/TestTaskContext.java
@@ -74,7 +74,8 @@ public class TestTaskContext<S, T> implements TaskContext<S, T> {
public TestTaskContext(long memoryInBytes) {
this.memoryManager = new MemoryManager(memoryInBytes, 1, 32 * 1024, MemoryType.HEAP, true);
- this.taskManageInfo = new TaskManagerRuntimeInfo("localhost", new Configuration());
+ this.taskManageInfo = new TaskManagerRuntimeInfo(
+ "localhost", new Configuration(), System.getProperty("java.io.tmpdir"));
}
// --------------------------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/bf256c7f/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/BinaryOperatorTestBase.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/BinaryOperatorTestBase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/BinaryOperatorTestBase.java
index 2c3dcf1..6e9b817 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/BinaryOperatorTestBase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/BinaryOperatorTestBase.java
@@ -110,7 +110,8 @@ public class BinaryOperatorTestBase<S extends Function, IN, OUT> extends TestLog
this.owner = new DummyInvokable();
this.taskConfig = new TaskConfig(new Configuration());
this.executionConfig = executionConfig;
- this.taskManageInfo = new TaskManagerRuntimeInfo("localhost", new Configuration());
+ this.taskManageInfo = new TaskManagerRuntimeInfo(
+ "localhost", new Configuration(), System.getProperty("java.io.tmpdir"));
}
@Parameterized.Parameters
http://git-wip-us.apache.org/repos/asf/flink/blob/bf256c7f/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/DriverTestBase.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/DriverTestBase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/DriverTestBase.java
index 6381733..eb2a3a7 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/DriverTestBase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/DriverTestBase.java
@@ -114,7 +114,8 @@ public class DriverTestBase<S extends Function> extends TestLogger implements Ta
this.owner = new DummyInvokable();
this.taskConfig = new TaskConfig(new Configuration());
this.executionConfig = executionConfig;
- this.taskManageInfo = new TaskManagerRuntimeInfo("localhost", new Configuration());
+ this.taskManageInfo = new TaskManagerRuntimeInfo(
+ "localhost", new Configuration(), System.getProperty("java.io.tmpdir"));
}
@Parameterized.Parameters
http://git-wip-us.apache.org/repos/asf/flink/blob/bf256c7f/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/MockEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/MockEnvironment.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/MockEnvironment.java
index 31fd08c..b774b48 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/MockEnvironment.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/MockEnvironment.java
@@ -208,7 +208,10 @@ public class MockEnvironment implements Environment {
@Override
public TaskManagerRuntimeInfo getTaskManagerInfo() {
- return new TaskManagerRuntimeInfo("localhost", new UnmodifiableConfiguration(new Configuration()));
+ return new TaskManagerRuntimeInfo(
+ "localhost",
+ new UnmodifiableConfiguration(new Configuration()),
+ System.getProperty("java.io.tmpdir"));
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/bf256c7f/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/UnaryOperatorTestBase.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/UnaryOperatorTestBase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/UnaryOperatorTestBase.java
index 77e18c6..50bb1ee 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/UnaryOperatorTestBase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/UnaryOperatorTestBase.java
@@ -115,7 +115,8 @@ public class UnaryOperatorTestBase<S extends Function, IN, OUT> extends TestLogg
this.executionConfig = executionConfig;
this.comparators = new ArrayList<TypeComparator<IN>>(2);
- this.taskManageInfo = new TaskManagerRuntimeInfo("localhost", new Configuration());
+ this.taskManageInfo = new TaskManagerRuntimeInfo(
+ "localhost", new Configuration(), System.getProperty("java.io.tmpdir"));
}
@Parameterized.Parameters
http://git-wip-us.apache.org/repos/asf/flink/blob/bf256c7f/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskAsyncCallTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskAsyncCallTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskAsyncCallTest.java
index ae05ae9..e1f551c 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskAsyncCallTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskAsyncCallTest.java
@@ -171,7 +171,7 @@ public class TaskAsyncCallTest {
new FiniteDuration(60, TimeUnit.SECONDS),
libCache,
mock(FileCache.class),
- new TaskManagerRuntimeInfo("localhost", new Configuration()),
+ new TaskManagerRuntimeInfo("localhost", new Configuration(), System.getProperty("java.io.tmpdir")),
mock(TaskMetricGroup.class));
}
http://git-wip-us.apache.org/repos/asf/flink/blob/bf256c7f/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskCancelTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskCancelTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskCancelTest.java
index 09dd817..fc5a4a3 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskCancelTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskCancelTest.java
@@ -18,7 +18,6 @@
package org.apache.flink.runtime.taskmanager;
-import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.JobID;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
@@ -41,7 +40,9 @@ import org.apache.flink.runtime.testingUtils.TestingCluster;
import org.apache.flink.runtime.testingUtils.TestingUtils;
import org.apache.flink.runtime.testutils.JobManagerActorTestUtils;
import org.apache.flink.types.IntValue;
+
import org.junit.Test;
+
import scala.concurrent.Await;
import scala.concurrent.Future;
import scala.concurrent.duration.FiniteDuration;
@@ -247,10 +248,11 @@ public class TaskCancelTest {
@Override
public void invoke() throws Exception {
UnionInputGate union = new UnionInputGate(getEnvironment().getAllInputGates());
- RecordReader<IntValue> reader = new RecordReader<>(union, IntValue.class);
+ RecordReader<IntValue> reader = new RecordReader<>(
+ union, IntValue.class, getEnvironment().getTaskManagerInfo().getTmpDirectories());
- while (reader.next() != null) {
- }
+ //noinspection StatementWithEmptyBody
+ while (reader.next() != null) {}
}
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/bf256c7f/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java
index f237c87..1762a7a 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java
@@ -623,7 +623,7 @@ public class TaskTest {
new FiniteDuration(60, TimeUnit.SECONDS),
libCache,
mock(FileCache.class),
- new TaskManagerRuntimeInfo("localhost", new Configuration()),
+ new TaskManagerRuntimeInfo("localhost", new Configuration(), System.getProperty("java.io.tmpdir")),
mock(TaskMetricGroup.class));
}
http://git-wip-us.apache.org/repos/asf/flink/blob/bf256c7f/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/Tasks.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/Tasks.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/Tasks.scala
index d871c3d..87c123a 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/Tasks.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/Tasks.scala
@@ -64,7 +64,11 @@ object Tasks {
class Forwarder extends AbstractInvokable {
override def invoke(): Unit = {
- val reader = new RecordReader[IntValue](getEnvironment.getInputGate(0), classOf[IntValue])
+ val reader = new RecordReader[IntValue](
+ getEnvironment.getInputGate(0),
+ classOf[IntValue],
+ getEnvironment.getTaskManagerInfo.getTmpDirectories)
+
val writer = new RecordWriter[IntValue](getEnvironment.getWriter(0))
try {
@@ -88,7 +92,10 @@ object Tasks {
class Receiver extends AbstractInvokable {
override def invoke(): Unit = {
- val reader = new RecordReader[IntValue](getEnvironment.getInputGate(0), classOf[IntValue])
+ val reader = new RecordReader[IntValue](
+ getEnvironment.getInputGate(0),
+ classOf[IntValue],
+ getEnvironment.getTaskManagerInfo.getTmpDirectories)
val i1 = reader.next()
val i2 = reader.next()
@@ -140,7 +147,10 @@ object Tasks {
class AgnosticReceiver extends AbstractInvokable {
override def invoke(): Unit = {
- val reader= new RecordReader[IntValue](getEnvironment.getInputGate(0), classOf[IntValue])
+ val reader= new RecordReader[IntValue](
+ getEnvironment.getInputGate(0),
+ classOf[IntValue],
+ getEnvironment.getTaskManagerInfo.getTmpDirectories)
while(reader.next() != null){}
}
@@ -149,8 +159,15 @@ object Tasks {
class AgnosticBinaryReceiver extends AbstractInvokable {
override def invoke(): Unit = {
- val reader1 = new RecordReader[IntValue](getEnvironment.getInputGate(0), classOf[IntValue])
- val reader2 = new RecordReader[IntValue](getEnvironment.getInputGate(1), classOf[IntValue])
+ val reader1 = new RecordReader[IntValue](
+ getEnvironment.getInputGate(0),
+ classOf[IntValue],
+ getEnvironment.getTaskManagerInfo.getTmpDirectories)
+
+ val reader2 = new RecordReader[IntValue](
+ getEnvironment.getInputGate(1),
+ classOf[IntValue],
+ getEnvironment.getTaskManagerInfo.getTmpDirectories)
while(reader1.next() != null){}
while(reader2.next() != null){}
@@ -162,9 +179,20 @@ object Tasks {
override def invoke(): Unit = {
val env = getEnvironment
- val reader1 = new RecordReader[IntValue](env.getInputGate(0), classOf[IntValue])
- val reader2 = new RecordReader[IntValue](env.getInputGate(1), classOf[IntValue])
- val reader3 = new RecordReader[IntValue](env.getInputGate(2), classOf[IntValue])
+ val reader1 = new RecordReader[IntValue](
+ env.getInputGate(0),
+ classOf[IntValue],
+ getEnvironment.getTaskManagerInfo.getTmpDirectories)
+
+ val reader2 = new RecordReader[IntValue](
+ env.getInputGate(1),
+ classOf[IntValue],
+ getEnvironment.getTaskManagerInfo.getTmpDirectories)
+
+ val reader3 = new RecordReader[IntValue](
+ env.getInputGate(2),
+ classOf[IntValue],
+ getEnvironment.getTaskManagerInfo.getTmpDirectories)
while(reader1.next() != null){}
while(reader2.next() != null){}
http://git-wip-us.apache.org/repos/asf/flink/blob/bf256c7f/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamInputProcessor.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamInputProcessor.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamInputProcessor.java
index 1dde85b..ab69ab7 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamInputProcessor.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamInputProcessor.java
@@ -117,7 +117,8 @@ public class StreamInputProcessor<IN> {
this.recordDeserializers = new SpillingAdaptiveSpanningRecordDeserializer[inputGate.getNumberOfInputChannels()];
for (int i = 0; i < recordDeserializers.length; i++) {
- recordDeserializers[i] = new SpillingAdaptiveSpanningRecordDeserializer<DeserializationDelegate<StreamElement>>();
+ recordDeserializers[i] = new SpillingAdaptiveSpanningRecordDeserializer<>(
+ ioManager.getSpillingDirectoriesPaths());
}
watermarks = new long[inputGate.getNumberOfInputChannels()];
http://git-wip-us.apache.org/repos/asf/flink/blob/bf256c7f/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamTwoInputProcessor.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamTwoInputProcessor.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamTwoInputProcessor.java
index 07ada23..733e7fb 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamTwoInputProcessor.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamTwoInputProcessor.java
@@ -139,7 +139,8 @@ public class StreamTwoInputProcessor<IN1, IN2> {
this.recordDeserializers = new SpillingAdaptiveSpanningRecordDeserializer[inputGate.getNumberOfInputChannels()];
for (int i = 0; i < recordDeserializers.length; i++) {
- recordDeserializers[i] = new SpillingAdaptiveSpanningRecordDeserializer<DeserializationDelegate<StreamElement>>();
+ recordDeserializers[i] = new SpillingAdaptiveSpanningRecordDeserializer<>(
+ ioManager.getSpillingDirectoriesPaths());
}
// determine which unioned channels belong to input 1 and which belong to input 2
http://git-wip-us.apache.org/repos/asf/flink/blob/bf256c7f/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamMockEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamMockEnvironment.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamMockEnvironment.java
index c62c881..b2d0196 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamMockEnvironment.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamMockEnvironment.java
@@ -23,7 +23,6 @@ import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.TaskInfo;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.UnmodifiableConfiguration;
import org.apache.flink.core.fs.Path;
import org.apache.flink.core.memory.MemorySegmentFactory;
import org.apache.flink.metrics.groups.TaskMetricGroup;
@@ -50,6 +49,7 @@ import org.apache.flink.runtime.plugable.DeserializationDelegate;
import org.apache.flink.runtime.plugable.NonReusingDeserializationDelegate;
import org.apache.flink.runtime.state.StateHandle;
import org.apache.flink.runtime.taskmanager.TaskManagerRuntimeInfo;
+
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
@@ -303,7 +303,7 @@ public class StreamMockEnvironment implements Environment {
@Override
public TaskManagerRuntimeInfo getTaskManagerInfo() {
- return new TaskManagerRuntimeInfo("localhost", new UnmodifiableConfiguration(new Configuration()));
+ return new TaskManagerRuntimeInfo("localhost", new Configuration(), System.getProperty("java.io.tmpdir"));
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/bf256c7f/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 deda82f..f054e18 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
@@ -158,7 +158,7 @@ public class StreamTaskTest {
new FiniteDuration(60, TimeUnit.SECONDS),
libCache,
mock(FileCache.class),
- new TaskManagerRuntimeInfo("localhost", new Configuration()),
+ new TaskManagerRuntimeInfo("localhost", new Configuration(), System.getProperty("java.io.tmpdir")),
mock(TaskMetricGroup.class));
}
http://git-wip-us.apache.org/repos/asf/flink/blob/bf256c7f/flink-tests/src/test/java/org/apache/flink/test/runtime/NetworkStackThroughputITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/runtime/NetworkStackThroughputITCase.java b/flink-tests/src/test/java/org/apache/flink/test/runtime/NetworkStackThroughputITCase.java
index 06df46f..5506f55 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/runtime/NetworkStackThroughputITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/runtime/NetworkStackThroughputITCase.java
@@ -18,27 +18,28 @@
package org.apache.flink.test.runtime;
-import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.io.IOReadableWritable;
import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.core.memory.DataOutputView;
import org.apache.flink.runtime.io.network.api.reader.RecordReader;
import org.apache.flink.runtime.io.network.api.writer.RecordWriter;
-import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.DistributionPattern;
import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup;
+import org.apache.flink.test.util.JavaProgramTestBase;
+
import org.junit.Ignore;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.Arrays;
import java.util.concurrent.TimeUnit;
-import org.apache.flink.api.common.JobExecutionResult;
-import org.apache.flink.test.util.JavaProgramTestBase;
@Ignore
public class NetworkStackThroughputITCase {
@@ -195,7 +196,11 @@ public class NetworkStackThroughputITCase {
@Override
public void invoke() throws Exception {
- RecordReader<SpeedTestRecord> reader = new RecordReader<>(getEnvironment().getInputGate(0), SpeedTestRecord.class);
+ RecordReader<SpeedTestRecord> reader = new RecordReader<>(
+ getEnvironment().getInputGate(0),
+ SpeedTestRecord.class,
+ getEnvironment().getTaskManagerInfo().getTmpDirectories());
+
RecordWriter<SpeedTestRecord> writer = new RecordWriter<>(getEnvironment().getWriter(0));
try {
@@ -215,7 +220,10 @@ public class NetworkStackThroughputITCase {
@Override
public void invoke() throws Exception {
- RecordReader<SpeedTestRecord> reader = new RecordReader<>(getEnvironment().getInputGate(0), SpeedTestRecord.class);
+ RecordReader<SpeedTestRecord> reader = new RecordReader<>(
+ getEnvironment().getInputGate(0),
+ SpeedTestRecord.class,
+ getEnvironment().getTaskManagerInfo().getTmpDirectories());
try {
boolean isSlow = getTaskConfiguration().getBoolean(IS_SLOW_RECEIVER_CONFIG_KEY, false);