Repository: flink Updated Branches: refs/heads/master 0b73b4387 -> 441ebf1ff
[hotfix] Fix TEZ task contect to reflect updated interface. Project: http://git-wip-us.apache.org/repos/asf/flink/repo Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/441ebf1f Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/441ebf1f Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/441ebf1f Branch: refs/heads/master Commit: 441ebf1ff381a6b4d9b6b289766d814195e560cb Parents: 0b73b43 Author: Stephan Ewen <se...@apache.org> Authored: Thu Aug 6 19:31:14 2015 +0200 Committer: Stephan Ewen <se...@apache.org> Committed: Thu Aug 6 19:36:07 2015 +0200 ---------------------------------------------------------------------- .../main/java/org/apache/flink/tez/runtime/TezTask.java | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/flink/blob/441ebf1f/flink-staging/flink-tez/src/main/java/org/apache/flink/tez/runtime/TezTask.java ---------------------------------------------------------------------- diff --git a/flink-staging/flink-tez/src/main/java/org/apache/flink/tez/runtime/TezTask.java b/flink-staging/flink-tez/src/main/java/org/apache/flink/tez/runtime/TezTask.java index 90df992..47fbad7 100644 --- a/flink-staging/flink-tez/src/main/java/org/apache/flink/tez/runtime/TezTask.java +++ b/flink-staging/flink-tez/src/main/java/org/apache/flink/tez/runtime/TezTask.java @@ -21,11 +21,13 @@ package org.apache.flink.tez.runtime; import com.google.common.base.Preconditions; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; + import org.apache.flink.api.common.ExecutionConfig; import org.apache.flink.api.common.distributions.DataDistribution; import org.apache.flink.api.common.functions.GroupCombineFunction; import org.apache.flink.api.common.functions.Function; import org.apache.flink.api.common.functions.util.FunctionUtils; +import org.apache.flink.api.common.functions.util.RuntimeUDFContext; import org.apache.flink.api.common.typeutils.TypeComparator; import org.apache.flink.api.common.typeutils.TypeComparatorFactory; import org.apache.flink.api.common.typeutils.TypeSerializer; @@ -39,10 +41,10 @@ import org.apache.flink.runtime.operators.PactTaskContext; import org.apache.flink.runtime.operators.shipping.ShipStrategyType; import org.apache.flink.runtime.operators.sort.CombiningUnilateralSortMerger; import org.apache.flink.runtime.operators.sort.UnilateralSortMerger; -import org.apache.flink.api.common.functions.util.RuntimeUDFContext; import org.apache.flink.runtime.operators.util.CloseableInputProvider; import org.apache.flink.runtime.operators.util.LocalStrategy; import org.apache.flink.runtime.operators.util.TaskConfig; +import org.apache.flink.runtime.taskmanager.TaskManagerRuntimeInfo; import org.apache.flink.tez.runtime.input.TezReaderIterator; import org.apache.flink.tez.runtime.output.TezChannelSelector; import org.apache.flink.tez.runtime.output.TezOutputEmitter; @@ -51,6 +53,7 @@ import org.apache.flink.tez.util.DummyInvokable; import org.apache.flink.util.Collector; import org.apache.flink.util.InstantiationUtil; import org.apache.flink.util.MutableObjectIterator; + import org.apache.tez.runtime.library.api.KeyValueReader; import org.apache.tez.runtime.library.api.KeyValueWriter; @@ -270,6 +273,11 @@ public class TezTask<S extends Function,OT> implements PactTaskContext<S, OT> { } @Override + public TaskManagerRuntimeInfo getTaskManagerInfo() { + return new TaskManagerRuntimeInfo("localhost", new Configuration()); + } + + @Override public <X> MutableObjectIterator<X> getInput(int index) { if (index < 0 || index > this.driver.getNumberOfInputs()) { throw new IndexOutOfBoundsException();