You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by fh...@apache.org on 2016/02/12 12:29:30 UTC
[05/50] [abbrv] flink git commit: [hotfix] Cleanup routing of records
in OperatorChain
[hotfix] Cleanup routing of records in OperatorChain
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/28c6254e
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/28c6254e
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/28c6254e
Branch: refs/heads/tableOnCalcite
Commit: 28c6254ee385fe746e868a81b2207bf66b552174
Parents: e9c83ea
Author: Stephan Ewen <se...@apache.org>
Authored: Mon Feb 8 16:14:00 2016 +0100
Committer: Stephan Ewen <se...@apache.org>
Committed: Mon Feb 8 20:36:35 2016 +0100
----------------------------------------------------------------------
.../BroadcastOutputSelectorWrapper.java | 45 -------
.../api/collector/selector/DirectedOutput.java | 130 +++++++++++++++++++
.../selector/DirectedOutputSelectorWrapper.java | 97 --------------
.../selector/OutputSelectorWrapper.java | 9 +-
.../selector/OutputSelectorWrapperFactory.java | 33 -----
.../flink/streaming/api/graph/StreamConfig.java | 20 +--
.../flink/streaming/api/graph/StreamNode.java | 10 +-
.../api/graph/StreamingJobGraphGenerator.java | 2 +-
.../streaming/runtime/io/CollectorWrapper.java | 61 ---------
.../streaming/runtime/tasks/OperatorChain.java | 84 ++++++++++--
.../flink/streaming/api/OutputSplitterTest.java | 2 +-
.../runtime/tasks/StreamTaskTestHarness.java | 9 +-
12 files changed, 225 insertions(+), 277 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/28c6254e/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/collector/selector/BroadcastOutputSelectorWrapper.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/collector/selector/BroadcastOutputSelectorWrapper.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/collector/selector/BroadcastOutputSelectorWrapper.java
deleted file mode 100644
index 7034b11..0000000
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/collector/selector/BroadcastOutputSelectorWrapper.java
+++ /dev/null
@@ -1,45 +0,0 @@
-/*
- * 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 regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.streaming.api.collector.selector;
-
-import java.util.ArrayList;
-
-import org.apache.flink.streaming.api.graph.StreamEdge;
-import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
-import org.apache.flink.util.Collector;
-
-public class BroadcastOutputSelectorWrapper<OUT> implements OutputSelectorWrapper<OUT> {
-
- private static final long serialVersionUID = 1L;
-
- private final ArrayList<Collector<StreamRecord<OUT>>> outputs;
-
- public BroadcastOutputSelectorWrapper() {
- outputs = new ArrayList<Collector<StreamRecord<OUT>>>();
- }
-
- @Override
- public void addCollector(Collector<StreamRecord<OUT>> output, StreamEdge edge) {
- outputs.add(output);
- }
-
- @Override
- public Iterable<Collector<StreamRecord<OUT>>> getSelectedOutputs(OUT record) {
- return outputs;
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/28c6254e/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/collector/selector/DirectedOutput.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/collector/selector/DirectedOutput.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/collector/selector/DirectedOutput.java
new file mode 100644
index 0000000..52c50b3
--- /dev/null
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/collector/selector/DirectedOutput.java
@@ -0,0 +1,130 @@
+/*
+ * 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 regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.api.collector.selector;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.streaming.api.graph.StreamEdge;
+import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+
+public class DirectedOutput<OUT> implements Output<StreamRecord<OUT>> {
+
+ private final OutputSelector<OUT>[] outputSelectors;
+
+ private final Output<StreamRecord<OUT>>[] selectAllOutputs;
+
+ private final HashMap<String, Output<StreamRecord<OUT>>[]> outputMap;
+
+ private final Output<StreamRecord<OUT>>[] allOutputs;
+
+
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ public DirectedOutput(
+ List<OutputSelector<OUT>> outputSelectors,
+ List<Tuple2<Output<StreamRecord<OUT>>, StreamEdge>> outputs)
+ {
+ this.outputSelectors = outputSelectors.toArray(new OutputSelector[outputSelectors.size()]);
+
+ this.allOutputs = new Output[outputs.size()];
+ for (int i = 0; i < outputs.size(); i++) {
+ allOutputs[i] = outputs.get(i).f0;
+ }
+
+
+ HashSet<Output<StreamRecord<OUT>>> selectAllOutputs = new HashSet<Output<StreamRecord<OUT>>>();
+ HashMap<String, ArrayList<Output<StreamRecord<OUT>>>> outputMap = new HashMap<String, ArrayList<Output<StreamRecord<OUT>>>>();
+
+ for (Tuple2<Output<StreamRecord<OUT>>, StreamEdge> outputPair : outputs) {
+ final Output<StreamRecord<OUT>> output = outputPair.f0;
+ final StreamEdge edge = outputPair.f1;
+
+ List<String> selectedNames = edge.getSelectedNames();
+
+ if (selectedNames.isEmpty()) {
+ selectAllOutputs.add(output);
+ }
+ else {
+ for (String selectedName : selectedNames) {
+ if (!outputMap.containsKey(selectedName)) {
+ outputMap.put(selectedName, new ArrayList<Output<StreamRecord<OUT>>>());
+ outputMap.get(selectedName).add(output);
+ }
+ else {
+ if (!outputMap.get(selectedName).contains(output)) {
+ outputMap.get(selectedName).add(output);
+ }
+ }
+ }
+ }
+ }
+
+ this.selectAllOutputs = selectAllOutputs.toArray(new Output[selectAllOutputs.size()]);
+
+ this.outputMap = new HashMap<>();
+ for (Map.Entry<String, ArrayList<Output<StreamRecord<OUT>>>> entry : outputMap.entrySet()) {
+ Output<StreamRecord<OUT>>[] arr = entry.getValue().toArray(new Output[entry.getValue().size()]);
+ this.outputMap.put(entry.getKey(), arr);
+ }
+ }
+
+
+ @Override
+ public void emitWatermark(Watermark mark) {
+ for (Output<StreamRecord<OUT>> out : allOutputs) {
+ out.emitWatermark(mark);
+ }
+ }
+
+ @Override
+ public void collect(StreamRecord<OUT> record) {
+ Set<Output<StreamRecord<OUT>>> selectedOutputs = new HashSet<Output<StreamRecord<OUT>>>(selectAllOutputs.length);
+ Collections.addAll(selectedOutputs, selectAllOutputs);
+
+ for (OutputSelector<OUT> outputSelector : outputSelectors) {
+ Iterable<String> outputNames = outputSelector.select(record.getValue());
+
+ for (String outputName : outputNames) {
+ Output<StreamRecord<OUT>>[] outputList = outputMap.get(outputName);
+ if (outputList != null) {
+ Collections.addAll(selectedOutputs, outputList);
+ }
+ }
+ }
+
+ for (Output<StreamRecord<OUT>> out : selectedOutputs) {
+ out.collect(record);
+ }
+ }
+
+ @Override
+ public void close() {
+ for (Output<StreamRecord<OUT>> out : allOutputs) {
+ out.close();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/28c6254e/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/collector/selector/DirectedOutputSelectorWrapper.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/collector/selector/DirectedOutputSelectorWrapper.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/collector/selector/DirectedOutputSelectorWrapper.java
deleted file mode 100644
index 84558fc..0000000
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/collector/selector/DirectedOutputSelectorWrapper.java
+++ /dev/null
@@ -1,97 +0,0 @@
-/*
- * 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 regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.streaming.api.collector.selector;
-
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Set;
-
-import org.apache.flink.streaming.api.graph.StreamEdge;
-import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
-import org.apache.flink.util.Collector;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class DirectedOutputSelectorWrapper<OUT> implements OutputSelectorWrapper<OUT> {
-
- private static final long serialVersionUID = 1L;
-
- private static final Logger LOG = LoggerFactory.getLogger(DirectedOutputSelectorWrapper.class);
-
- private List<OutputSelector<OUT>> outputSelectors;
-
- private HashMap<String, ArrayList<Collector<StreamRecord<OUT>>>> outputMap;
- private HashSet<Collector<StreamRecord<OUT>>> selectAllOutputs;
-
- public DirectedOutputSelectorWrapper(List<OutputSelector<OUT>> outputSelectors) {
- this.outputSelectors = outputSelectors;
- this.selectAllOutputs = new HashSet<Collector<StreamRecord<OUT>>>();
- this.outputMap = new HashMap<String, ArrayList<Collector<StreamRecord<OUT>>>>();
- }
-
- @Override
- public void addCollector(Collector<StreamRecord<OUT>> output, StreamEdge edge) {
- List<String> selectedNames = edge.getSelectedNames();
-
- if (selectedNames.isEmpty()) {
- selectAllOutputs.add(output);
- }
- else {
- for (String selectedName : selectedNames) {
- if (!outputMap.containsKey(selectedName)) {
- outputMap.put(selectedName, new ArrayList<Collector<StreamRecord<OUT>>>());
- outputMap.get(selectedName).add(output);
- }
- else {
- if (!outputMap.get(selectedName).contains(output)) {
- outputMap.get(selectedName).add(output);
- }
- }
- }
- }
- }
-
- @Override
- public Iterable<Collector<StreamRecord<OUT>>> getSelectedOutputs(OUT record) {
- Set<Collector<StreamRecord<OUT>>> selectedOutputs = new HashSet<Collector<StreamRecord<OUT>>>(selectAllOutputs);
-
- for (OutputSelector<OUT> outputSelector : outputSelectors) {
- Iterable<String> outputNames = outputSelector.select(record);
-
- for (String outputName : outputNames) {
- List<Collector<StreamRecord<OUT>>> outputList = outputMap.get(outputName);
-
- try {
- selectedOutputs.addAll(outputList);
- } catch (NullPointerException e) {
- if (LOG.isErrorEnabled()) {
- String format = String.format(
- "Cannot emit because no output is selected with the name: %s",
- outputName);
- LOG.error(format);
- }
- }
- }
- }
-
- return selectedOutputs;
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/28c6254e/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/collector/selector/OutputSelectorWrapper.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/collector/selector/OutputSelectorWrapper.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/collector/selector/OutputSelectorWrapper.java
index f25c995..971e42b 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/collector/selector/OutputSelectorWrapper.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/collector/selector/OutputSelectorWrapper.java
@@ -19,14 +19,7 @@ package org.apache.flink.streaming.api.collector.selector;
import java.io.Serializable;
-import org.apache.flink.streaming.api.graph.StreamEdge;
-import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
-import org.apache.flink.util.Collector;
-
public interface OutputSelectorWrapper<OUT> extends Serializable {
- public void addCollector(Collector<StreamRecord<OUT>> output, StreamEdge edge);
-
- public Iterable<Collector<StreamRecord<OUT>>> getSelectedOutputs(OUT record);
-
+ void sendOutputs(OUT record);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/28c6254e/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/collector/selector/OutputSelectorWrapperFactory.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/collector/selector/OutputSelectorWrapperFactory.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/collector/selector/OutputSelectorWrapperFactory.java
deleted file mode 100644
index dca2ede..0000000
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/collector/selector/OutputSelectorWrapperFactory.java
+++ /dev/null
@@ -1,33 +0,0 @@
-/*
- * 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 regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.streaming.api.collector.selector;
-
-import java.util.List;
-
-public class OutputSelectorWrapperFactory {
-
- @SuppressWarnings({ "rawtypes", "unchecked" })
- public static OutputSelectorWrapper<?> create(List<OutputSelector<?>> outputSelectors) {
- if (outputSelectors.size() == 0) {
- return new BroadcastOutputSelectorWrapper();
- } else {
- return new DirectedOutputSelectorWrapper(outputSelectors);
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/28c6254e/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamConfig.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamConfig.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamConfig.java
index 7a07c79..311b7fb 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamConfig.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamConfig.java
@@ -20,6 +20,7 @@ package org.apache.flink.streaming.api.graph;
import java.io.IOException;
import java.io.Serializable;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -29,7 +30,7 @@ import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.util.ClassLoaderUtil;
import org.apache.flink.streaming.api.CheckpointingMode;
-import org.apache.flink.streaming.api.collector.selector.OutputSelectorWrapper;
+import org.apache.flink.streaming.api.collector.selector.OutputSelector;
import org.apache.flink.streaming.api.operators.StreamOperator;
import org.apache.flink.runtime.state.AbstractStateBackend;
import org.apache.flink.streaming.runtime.tasks.StreamTaskException;
@@ -38,7 +39,7 @@ import org.apache.flink.util.InstantiationUtil;
public class StreamConfig implements Serializable {
private static final long serialVersionUID = 1L;
-
+
// ------------------------------------------------------------------------
// Config Keys
// ------------------------------------------------------------------------
@@ -191,19 +192,22 @@ public class StreamConfig implements Serializable {
}
}
- public void setOutputSelectorWrapper(OutputSelectorWrapper<?> outputSelectorWrapper) {
+ public void setOutputSelectors(List<OutputSelector<?>> outputSelectors) {
try {
- InstantiationUtil.writeObjectToConfig(outputSelectorWrapper, this.config, OUTPUT_SELECTOR_WRAPPER);
+ InstantiationUtil.writeObjectToConfig(outputSelectors, this.config, OUTPUT_SELECTOR_WRAPPER);
} catch (IOException e) {
- throw new StreamTaskException("Cannot serialize OutputSelectorWrapper.", e);
+ throw new StreamTaskException("Could not serialize output selectors", e);
}
}
- public <T> OutputSelectorWrapper<T> getOutputSelectorWrapper(ClassLoader cl) {
+ public <T> List<OutputSelector<T>> getOutputSelectors(ClassLoader userCodeClassloader) {
try {
- return InstantiationUtil.readObjectFromConfig(this.config, OUTPUT_SELECTOR_WRAPPER, cl);
+ List<OutputSelector<T>> selectors =
+ InstantiationUtil.readObjectFromConfig(this.config, OUTPUT_SELECTOR_WRAPPER, userCodeClassloader);
+ return selectors == null ? Collections.<OutputSelector<T>>emptyList() : selectors;
+
} catch (Exception e) {
- throw new StreamTaskException("Cannot deserialize and instantiate OutputSelectorWrapper.", e);
+ throw new StreamTaskException("Could not read output selectors", e);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/28c6254e/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamNode.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamNode.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamNode.java
index 0a612f3..3e06037 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamNode.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamNode.java
@@ -26,15 +26,11 @@ import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
import org.apache.flink.streaming.api.collector.selector.OutputSelector;
-import org.apache.flink.streaming.api.collector.selector.OutputSelectorWrapper;
-import org.apache.flink.streaming.api.collector.selector.OutputSelectorWrapperFactory;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.operators.StreamOperator;
/**
- * Class representing the operators in the streaming programs, with all their
- * properties.
- *
+ * Class representing the operators in the streaming programs, with all their properties.
*/
public class StreamNode implements Serializable {
@@ -168,10 +164,6 @@ public class StreamNode implements Serializable {
return outputSelectors;
}
- public OutputSelectorWrapper<?> getOutputSelectorWrapper() {
- return OutputSelectorWrapperFactory.create(getOutputSelectors());
- }
-
public void addOutputSelector(OutputSelector<?> outputSelector) {
this.outputSelectors.add(outputSelector);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/28c6254e/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
index c0d2856..c810e47 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
@@ -310,7 +310,7 @@ public class StreamingJobGraphGenerator {
config.setTypeSerializerOut(vertex.getTypeSerializerOut());
config.setStreamOperator(vertex.getOperator());
- config.setOutputSelectorWrapper(vertex.getOutputSelectorWrapper());
+ config.setOutputSelectors(vertex.getOutputSelectors());
config.setNumberOfOutputs(nonChainableOutputs.size());
config.setNonChainedOutputs(nonChainableOutputs);
http://git-wip-us.apache.org/repos/asf/flink/blob/28c6254e/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CollectorWrapper.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CollectorWrapper.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CollectorWrapper.java
deleted file mode 100644
index 01e997d..0000000
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CollectorWrapper.java
+++ /dev/null
@@ -1,61 +0,0 @@
-/*
- * 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 regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.streaming.runtime.io;
-
-import org.apache.flink.streaming.api.collector.selector.OutputSelectorWrapper;
-import org.apache.flink.streaming.api.graph.StreamEdge;
-import org.apache.flink.streaming.api.operators.Output;
-import org.apache.flink.streaming.api.watermark.Watermark;
-import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
-import org.apache.flink.util.Collector;
-
-import java.util.ArrayList;
-
-public class CollectorWrapper<OUT> implements Output<StreamRecord<OUT>> {
-
- private OutputSelectorWrapper<OUT> outputSelectorWrapper;
-
- private ArrayList<Output<StreamRecord<OUT>>> allOutputs;
-
- public CollectorWrapper(OutputSelectorWrapper<OUT> outputSelectorWrapper) {
- this.outputSelectorWrapper = outputSelectorWrapper;
- allOutputs = new ArrayList<Output<StreamRecord<OUT>>>();
- }
-
- public void addCollector(Output<StreamRecord<OUT>> output, StreamEdge edge) {
- outputSelectorWrapper.addCollector(output, edge);
- allOutputs.add(output);
- }
-
- @Override
- public void collect(StreamRecord<OUT> record) {
- for (Collector<StreamRecord<OUT>> output : outputSelectorWrapper.getSelectedOutputs(record.getValue())) {
- output.collect(record);
- }
- }
-
- @Override
- public void emitWatermark(Watermark mark) {
- for (Output<?> output : allOutputs) {
- output.emitWatermark(mark);
- }
- }
-
- @Override
- public void close() {}
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/28c6254e/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/OperatorChain.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/OperatorChain.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/OperatorChain.java
index 125279c..5313bc9 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/OperatorChain.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/OperatorChain.java
@@ -24,15 +24,16 @@ import java.util.List;
import java.util.Map;
import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.runtime.accumulators.AccumulatorRegistry;
import org.apache.flink.runtime.execution.Environment;
import org.apache.flink.runtime.io.network.api.CheckpointBarrier;
import org.apache.flink.runtime.io.network.api.writer.ResultPartitionWriter;
import org.apache.flink.runtime.plugable.SerializationDelegate;
+import org.apache.flink.streaming.api.collector.selector.DirectedOutput;
+import org.apache.flink.streaming.api.collector.selector.OutputSelector;
import org.apache.flink.streaming.api.watermark.Watermark;
-import org.apache.flink.streaming.runtime.io.CollectorWrapper;
import org.apache.flink.streaming.runtime.io.RecordWriterOutput;
-import org.apache.flink.streaming.api.collector.selector.OutputSelectorWrapper;
import org.apache.flink.streaming.api.graph.StreamConfig;
import org.apache.flink.streaming.api.graph.StreamEdge;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
@@ -45,6 +46,13 @@ import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+/**
+ * The {@code OperatorChain} contains all operators that are executed as one chain within a single
+ * {@link StreamTask}.
+ *
+ * @param <OUT> The type of elements accepted by the chain, i.e., the input type of the chain's
+ * head operator.
+ */
public class OperatorChain<OUT> {
private static final Logger LOG = LoggerFactory.getLogger(OperatorChain.class);
@@ -182,15 +190,14 @@ public class OperatorChain<OUT> {
Map<StreamEdge, RecordWriterOutput<?>> streamOutputs,
List<StreamOperator<?>> allOperators)
{
- // We create a wrapper that will encapsulate the chained operators and network outputs
- OutputSelectorWrapper<T> outputSelectorWrapper = operatorConfig.getOutputSelectorWrapper(userCodeClassloader);
- CollectorWrapper<T> wrapper = new CollectorWrapper<T>(outputSelectorWrapper);
-
+ List<Tuple2<Output<StreamRecord<T>>, StreamEdge>> allOutputs = new ArrayList<>(4);
+
// create collectors for the network outputs
for (StreamEdge outputEdge : operatorConfig.getNonChainedOutputs(userCodeClassloader)) {
@SuppressWarnings("unchecked")
RecordWriterOutput<T> output = (RecordWriterOutput<T>) streamOutputs.get(outputEdge);
- wrapper.addCollector(output, outputEdge);
+
+ allOutputs.add(new Tuple2<Output<StreamRecord<T>>, StreamEdge>(output, outputEdge));
}
// Create collectors for the chained outputs
@@ -200,9 +207,37 @@ public class OperatorChain<OUT> {
Output<StreamRecord<T>> output = createChainedOperator(
containingTask, chainedOpConfig, chainedConfigs, userCodeClassloader, streamOutputs, allOperators);
- wrapper.addCollector(output, outputEdge);
+
+ allOutputs.add(new Tuple2<>(output, outputEdge));
+ }
+
+ // if there are multiple outputs, or the outputs are directed, we need to
+ // wrap them as one output
+
+ List<OutputSelector<T>> selectors = operatorConfig.getOutputSelectors(userCodeClassloader);
+
+ if (selectors == null || selectors.isEmpty()) {
+ // simple path, no selector necessary
+ if (allOutputs.size() == 1) {
+ return allOutputs.get(0).f0;
+ }
+ else {
+ // send to N outputs. Note that this includes teh special case
+ // of sending to zero outputs
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ Output<StreamRecord<T>>[] asArray = new Output[allOutputs.size()];
+ for (int i = 0; i < allOutputs.size(); i++) {
+ asArray[i] = allOutputs.get(i).f0;
+ }
+
+ return new BroadcastingOutputCollector<T>(asArray);
+ }
+ }
+ else {
+ // selector present, more complex routing necessary
+ return new DirectedOutput<T>(selectors, allOutputs);
+
}
- return wrapper;
}
private static <IN, OUT> Output<StreamRecord<IN>> createChainedOperator(
@@ -309,7 +344,6 @@ public class OperatorChain<OUT> {
@Override
public void collect(StreamRecord<T> record) {
try {
-
StreamRecord<T> copy = new StreamRecord<>(serializer.copy(record.getValue()), record.getTimestamp());
operator.setKeyContextElement1(copy);
@@ -320,4 +354,34 @@ public class OperatorChain<OUT> {
}
}
}
+
+ private static final class BroadcastingOutputCollector<T> implements Output<StreamRecord<T>> {
+
+ private final Output<StreamRecord<T>>[] outputs;
+
+ public BroadcastingOutputCollector(Output<StreamRecord<T>>[] outputs) {
+ this.outputs = outputs;
+ }
+
+ @Override
+ public void emitWatermark(Watermark mark) {
+ for (Output<StreamRecord<T>> output : outputs) {
+ output.emitWatermark(mark);
+ }
+ }
+
+ @Override
+ public void collect(StreamRecord<T> record) {
+ for (Output<StreamRecord<T>> output : outputs) {
+ output.collect(record);
+ }
+ }
+
+ @Override
+ public void close() {
+ for (Output<StreamRecord<T>> output : outputs) {
+ output.close();
+ }
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/28c6254e/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/OutputSplitterTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/OutputSplitterTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/OutputSplitterTest.java
index 8525d37..5126d11 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/OutputSplitterTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/OutputSplitterTest.java
@@ -28,7 +28,7 @@ import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.util.StreamingMultipleProgramsTestBase;
import org.apache.flink.streaming.util.TestListResultSink;
-import org.apache.flink.streaming.util.TestStreamEnvironment;
+
import org.junit.Test;
public class OutputSplitterTest extends StreamingMultipleProgramsTestBase {
http://git-wip-us.apache.org/repos/asf/flink/blob/28c6254e/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTestHarness.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTestHarness.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTestHarness.java
index 2cca3ff..e32b304 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTestHarness.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTestHarness.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
@@ -15,6 +15,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.flink.streaming.runtime.tasks;
import org.apache.flink.api.common.ExecutionConfig;
@@ -26,7 +27,6 @@ import org.apache.flink.runtime.io.network.partition.consumer.StreamTestSingleIn
import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
import org.apache.flink.runtime.memory.MemoryManager;
import org.apache.flink.runtime.operators.testutils.MockInputSplitProvider;
-import org.apache.flink.streaming.api.collector.selector.BroadcastOutputSelectorWrapper;
import org.apache.flink.streaming.api.collector.selector.OutputSelector;
import org.apache.flink.streaming.api.graph.StreamConfig;
import org.apache.flink.streaming.api.graph.StreamEdge;
@@ -40,11 +40,11 @@ import org.apache.flink.util.InstantiationUtil;
import org.junit.Assert;
import java.io.IOException;
+import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;
-
/**
* Test harness for testing a {@link StreamTask}.
*
@@ -91,6 +91,7 @@ public class StreamTaskTestHarness<OUT> {
// input related methods only need to be implemented once, in generic form
protected int numInputGates;
protected int numInputChannelsPerGate;
+
@SuppressWarnings("rawtypes")
protected StreamTestSingleInputGate[] inputGates;
@@ -128,7 +129,7 @@ public class StreamTaskTestHarness<OUT> {
mockEnv.addOutput(outputList, outputStreamRecordSerializer);
- streamConfig.setOutputSelectorWrapper(new BroadcastOutputSelectorWrapper<Object>());
+ streamConfig.setOutputSelectors(Collections.<OutputSelector<?>>emptyList());
streamConfig.setNumberOfOutputs(1);
StreamOperator<OUT> dummyOperator = new AbstractStreamOperator<OUT>() {