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 2014/10/03 18:25:15 UTC
[19/23] git commit: [FLINK-1110] Fix mutable object safe mode for
data sources in collection-based execution
[FLINK-1110] Fix mutable object safe mode for data sources in collection-based execution
Project: http://git-wip-us.apache.org/repos/asf/incubator-flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-flink/commit/3cc9a289
Tree: http://git-wip-us.apache.org/repos/asf/incubator-flink/tree/3cc9a289
Diff: http://git-wip-us.apache.org/repos/asf/incubator-flink/diff/3cc9a289
Branch: refs/heads/master
Commit: 3cc9a289fb93c18085f251422446e9e987cd51fb
Parents: 54ede63
Author: Stephan Ewen <se...@apache.org>
Authored: Wed Oct 1 20:11:52 2014 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Fri Oct 3 16:22:34 2014 +0200
----------------------------------------------------------------------
.../apache/flink/api/common/operators/CollectionExecutor.java | 4 ++--
.../flink/api/common/operators/GenericDataSourceBase.java | 4 ++--
.../java/org/apache/flink/api/java/CollectionEnvironment.java | 6 ++++--
3 files changed, 8 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/3cc9a289/flink-core/src/main/java/org/apache/flink/api/common/operators/CollectionExecutor.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/operators/CollectionExecutor.java b/flink-core/src/main/java/org/apache/flink/api/common/operators/CollectionExecutor.java
index 5d30df4..e82d7cb 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/operators/CollectionExecutor.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/operators/CollectionExecutor.java
@@ -1,4 +1,4 @@
-/**
+/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
@@ -152,7 +152,7 @@ public class CollectionExecutor {
private <OUT> List<OUT> executeDataSource(GenericDataSourceBase<?, ?> source) throws Exception {
@SuppressWarnings("unchecked")
GenericDataSourceBase<OUT, ?> typedSource = (GenericDataSourceBase<OUT, ?>) source;
- return typedSource.executeOnCollections();
+ return typedSource.executeOnCollections(mutableObjectSafeMode);
}
private <IN, OUT> List<OUT> executeUnaryOperator(SingleInputOperator<?, ?, ?> operator, int superStep) throws Exception {
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/3cc9a289/flink-core/src/main/java/org/apache/flink/api/common/operators/GenericDataSourceBase.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/operators/GenericDataSourceBase.java b/flink-core/src/main/java/org/apache/flink/api/common/operators/GenericDataSourceBase.java
index 118b95c..ad1b2e4 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/operators/GenericDataSourceBase.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/operators/GenericDataSourceBase.java
@@ -176,7 +176,7 @@ public class GenericDataSourceBase<OUT, T extends InputFormat<OUT, ?>> extends O
// --------------------------------------------------------------------------------------------
- protected List<OUT> executeOnCollections() throws Exception {
+ protected List<OUT> executeOnCollections(boolean mutableObjectSafe) throws Exception {
@SuppressWarnings("unchecked")
InputFormat<OUT, InputSplit> inputFormat = (InputFormat<OUT, InputSplit>) this.formatWrapper.getUserCodeObject();
inputFormat.configure(this.parameters);
@@ -193,7 +193,7 @@ public class GenericDataSourceBase<OUT, T extends InputFormat<OUT, ?>> extends O
while (!inputFormat.reachedEnd()) {
OUT next = inputFormat.nextRecord(serializer.createInstance());
if (next != null) {
- result.add(next);
+ result.add(mutableObjectSafe ? serializer.copy(next) : next);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/3cc9a289/flink-java/src/main/java/org/apache/flink/api/java/CollectionEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/CollectionEnvironment.java b/flink-java/src/main/java/org/apache/flink/api/java/CollectionEnvironment.java
index 52cea4b..3e9ff66 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/CollectionEnvironment.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/CollectionEnvironment.java
@@ -1,4 +1,4 @@
-/**
+/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
@@ -24,11 +24,13 @@ import org.apache.flink.api.common.operators.CollectionExecutor;
public class CollectionEnvironment extends ExecutionEnvironment {
+ private boolean mutableObjectSafeMode = true;
+
@Override
public JobExecutionResult execute(String jobName) throws Exception {
Plan p = createProgramPlan(jobName);
- CollectionExecutor exec = new CollectionExecutor();
+ CollectionExecutor exec = new CollectionExecutor(mutableObjectSafeMode);
return exec.execute(p);
}