You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ga...@apache.org on 2021/10/27 15:53:05 UTC
[flink-ml] branch master updated: [FLINK-24644][iteration]
Introduce broadcast output to support broadcast events to all the
downstream operators
This is an automated email from the ASF dual-hosted git repository.
gaoyunhaii pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-ml.git
The following commit(s) were added to refs/heads/master by this push:
new 5b3e84b [FLINK-24644][iteration] Introduce broadcast output to support broadcast events to all the downstream operators
5b3e84b is described below
commit 5b3e84b23e8d914bc68841eb2f2b28f8c50f42c6
Author: Yun Gao <ga...@gmail.com>
AuthorDate: Thu Sep 23 16:24:15 2021 +0800
[FLINK-24644][iteration] Introduce broadcast output to support broadcast events to all the downstream operators
This closes #6.
---
flink-ml-iteration/pom.xml | 14 ++
.../flink/iteration/broadcast/BroadcastOutput.java | 34 +++
.../broadcast/BroadcastOutputFactory.java | 78 +++++++
.../broadcast/ChainingBroadcastOutput.java | 43 ++++
.../broadcast/CountingBroadcastOutput.java | 50 +++++
.../broadcast/OutputReflectionContext.java | 104 +++++++++
.../broadcast/RecordWriterBroadcastOutput.java | 46 ++++
.../flink/iteration/utils/ReflectionUtils.java | 52 +++++
.../iteration/broadcast/BroadcastOutputTest.java | 248 +++++++++++++++++++++
.../flink/iteration/utils/ReflectionUtilsTest.java | 49 ++++
10 files changed, 718 insertions(+)
diff --git a/flink-ml-iteration/pom.xml b/flink-ml-iteration/pom.xml
index 6dddfc4..88d3ef9 100644
--- a/flink-ml-iteration/pom.xml
+++ b/flink-ml-iteration/pom.xml
@@ -41,9 +41,23 @@ under the License.
<dependency>
<groupId>org.apache.flink</groupId>
+ <artifactId>flink-metrics-core</artifactId>
+ <version>${flink.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
<scope>provided</scope>
</dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-clients_${scala.binary.version}</artifactId>
+ <version>${flink.version}</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
</project>
diff --git a/flink-ml-iteration/src/main/java/org/apache/flink/iteration/broadcast/BroadcastOutput.java b/flink-ml-iteration/src/main/java/org/apache/flink/iteration/broadcast/BroadcastOutput.java
new file mode 100644
index 0000000..93f1b88
--- /dev/null
+++ b/flink-ml-iteration/src/main/java/org/apache/flink/iteration/broadcast/BroadcastOutput.java
@@ -0,0 +1,34 @@
+/*
+ * 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.iteration.broadcast;
+
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+import java.io.IOException;
+
+/** Wrapper output that broadcasts the records to all the downstream operators. */
+public interface BroadcastOutput<OUT> {
+
+ /**
+ * Broadcasts the record to all the downstream operators.
+ *
+ * @param record the record to broadcast.
+ */
+ void broadcastEmit(StreamRecord<OUT> record) throws IOException;
+}
diff --git a/flink-ml-iteration/src/main/java/org/apache/flink/iteration/broadcast/BroadcastOutputFactory.java b/flink-ml-iteration/src/main/java/org/apache/flink/iteration/broadcast/BroadcastOutputFactory.java
new file mode 100644
index 0000000..29fc3c7
--- /dev/null
+++ b/flink-ml-iteration/src/main/java/org/apache/flink/iteration/broadcast/BroadcastOutputFactory.java
@@ -0,0 +1,78 @@
+/*
+ * 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.iteration.broadcast;
+
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.runtime.io.network.api.writer.RecordWriter;
+import org.apache.flink.runtime.plugable.SerializationDelegate;
+import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.runtime.streamrecord.StreamElement;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.OutputTag;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/** Factory that creates the corresponding {@link BroadcastOutput} from the {@link Output}. */
+public class BroadcastOutputFactory {
+
+ /**
+ * Creates the wrapper broadcast output from {@code output}.
+ *
+ * @param output the original output.
+ * @param numRecordsOut the counter for the number of output record.
+ * @return the wrapped broadcast output.
+ */
+ public static <OUT> BroadcastOutput<OUT> createBroadcastOutput(
+ Output<StreamRecord<OUT>> output, Counter numRecordsOut) {
+
+ OutputReflectionContext outputReflectionContext = new OutputReflectionContext();
+
+ List<BroadcastOutput<OUT>> internalOutputs = new ArrayList<>();
+ if (outputReflectionContext.isBroadcastingOutput(output)) {
+ Output<StreamRecord<OUT>>[] rawOutputs =
+ outputReflectionContext.getBroadcastingInternalOutputs(output);
+ for (Output<StreamRecord<OUT>> rawOutput : rawOutputs) {
+ internalOutputs.add(
+ createInternalBroadcastOutput(rawOutput, outputReflectionContext));
+ }
+ } else {
+ internalOutputs.add(createInternalBroadcastOutput(output, outputReflectionContext));
+ }
+
+ return new CountingBroadcastOutput<>(numRecordsOut, internalOutputs);
+ }
+
+ private static <OUT> BroadcastOutput<OUT> createInternalBroadcastOutput(
+ Output<StreamRecord<OUT>> rawOutput, OutputReflectionContext outputReflectionContext) {
+
+ if (outputReflectionContext.isChainingOutput(rawOutput)) {
+ OutputTag<?> outputTag = outputReflectionContext.getChainingOutputTag(rawOutput);
+ return new ChainingBroadcastOutput<>(rawOutput, outputTag);
+ } else if (outputReflectionContext.isRecordWriterOutput(rawOutput)) {
+ RecordWriter<SerializationDelegate<StreamElement>> recordWriter =
+ outputReflectionContext.getRecordWriter(rawOutput);
+ TypeSerializer<StreamElement> typeSerializer =
+ outputReflectionContext.getRecordWriterTypeSerializer(rawOutput);
+ return new RecordWriterBroadcastOutput<>(recordWriter, typeSerializer);
+ } else {
+ throw new RuntimeException("Unknown output type: " + rawOutput.getClass());
+ }
+ }
+}
diff --git a/flink-ml-iteration/src/main/java/org/apache/flink/iteration/broadcast/ChainingBroadcastOutput.java b/flink-ml-iteration/src/main/java/org/apache/flink/iteration/broadcast/ChainingBroadcastOutput.java
new file mode 100644
index 0000000..a97476b
--- /dev/null
+++ b/flink-ml-iteration/src/main/java/org/apache/flink/iteration/broadcast/ChainingBroadcastOutput.java
@@ -0,0 +1,43 @@
+/*
+ * 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.iteration.broadcast;
+
+import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.OutputTag;
+
+/** The broadcast output corresponding to a chained output. */
+public class ChainingBroadcastOutput<OUT> implements BroadcastOutput<OUT> {
+ private final Output<StreamRecord<OUT>> rawOutput;
+ private final OutputTag outputTag;
+
+ ChainingBroadcastOutput(Output<StreamRecord<OUT>> rawOutput, OutputTag outputTag) {
+ this.rawOutput = rawOutput;
+ this.outputTag = outputTag;
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public void broadcastEmit(StreamRecord<OUT> record) {
+ if (outputTag == null) {
+ rawOutput.collect(record);
+ } else {
+ rawOutput.collect(outputTag, record);
+ }
+ }
+}
diff --git a/flink-ml-iteration/src/main/java/org/apache/flink/iteration/broadcast/CountingBroadcastOutput.java b/flink-ml-iteration/src/main/java/org/apache/flink/iteration/broadcast/CountingBroadcastOutput.java
new file mode 100644
index 0000000..08713b6
--- /dev/null
+++ b/flink-ml-iteration/src/main/java/org/apache/flink/iteration/broadcast/CountingBroadcastOutput.java
@@ -0,0 +1,50 @@
+/*
+ * 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.iteration.broadcast;
+
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+import java.io.IOException;
+import java.util.List;
+
+/**
+ * The intermediate broadcast output that wrappers a list of internal outputs. It will broadcast the
+ * records to all the internal outputs and increment the counter.
+ */
+public class CountingBroadcastOutput<OUT> implements BroadcastOutput<OUT> {
+
+ private final Counter numRecordsOut;
+ private final List<BroadcastOutput<OUT>> internalOutputs;
+
+ public CountingBroadcastOutput(
+ Counter numRecordsOut, List<BroadcastOutput<OUT>> internalOutputs) {
+ this.numRecordsOut = numRecordsOut;
+ this.internalOutputs = internalOutputs;
+ }
+
+ @Override
+ public void broadcastEmit(StreamRecord<OUT> record) throws IOException {
+ numRecordsOut.inc();
+
+ for (BroadcastOutput<OUT> internalOutput : internalOutputs) {
+ internalOutput.broadcastEmit(record);
+ }
+ }
+}
diff --git a/flink-ml-iteration/src/main/java/org/apache/flink/iteration/broadcast/OutputReflectionContext.java b/flink-ml-iteration/src/main/java/org/apache/flink/iteration/broadcast/OutputReflectionContext.java
new file mode 100644
index 0000000..6ce8210
--- /dev/null
+++ b/flink-ml-iteration/src/main/java/org/apache/flink/iteration/broadcast/OutputReflectionContext.java
@@ -0,0 +1,104 @@
+/*
+ * 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.iteration.broadcast;
+
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.iteration.utils.ReflectionUtils;
+import org.apache.flink.runtime.io.network.api.writer.RecordWriter;
+import org.apache.flink.runtime.plugable.SerializationDelegate;
+import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.runtime.io.RecordWriterOutput;
+import org.apache.flink.streaming.runtime.streamrecord.StreamElement;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.OutputTag;
+
+import java.lang.reflect.Field;
+
+/** The reflection utilities to parse the output and create the broadcast output. */
+public class OutputReflectionContext {
+ private final Class<?> broadcastingOutputClass;
+ private final Field broadcastingOutputsField;
+
+ private final Class<?> chainingOutputClass;
+ private final Field chainingOutputTagField;
+
+ private final Field recordWriterField;
+ private final Field recordWriterSerializationDelegateField;
+ private final Field serializationDelegateSerializerField;
+
+ public OutputReflectionContext() {
+ try {
+ this.broadcastingOutputClass =
+ Class.forName(
+ "org.apache.flink.streaming.runtime.tasks.BroadcastingOutputCollector");
+ this.broadcastingOutputsField =
+ ReflectionUtils.getClassField(broadcastingOutputClass, "outputs");
+
+ this.chainingOutputClass =
+ Class.forName("org.apache.flink.streaming.runtime.tasks.ChainingOutput");
+ this.chainingOutputTagField =
+ ReflectionUtils.getClassField(chainingOutputClass, "outputTag");
+
+ this.recordWriterField =
+ ReflectionUtils.getClassField(RecordWriterOutput.class, "recordWriter");
+ this.recordWriterSerializationDelegateField =
+ ReflectionUtils.getClassField(
+ RecordWriterOutput.class, "serializationDelegate");
+ this.serializationDelegateSerializerField =
+ ReflectionUtils.getClassField(SerializationDelegate.class, "serializer");
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to initialize the OutputReflectionContext", e);
+ }
+ }
+
+ public boolean isBroadcastingOutput(Output<?> rawOutput) {
+ return broadcastingOutputClass.isAssignableFrom(rawOutput.getClass());
+ }
+
+ public boolean isChainingOutput(Output<?> rawOutput) {
+ return chainingOutputClass.isAssignableFrom(rawOutput.getClass());
+ }
+
+ public boolean isRecordWriterOutput(Output<?> rawOutput) {
+ return RecordWriterOutput.class.isAssignableFrom(rawOutput.getClass());
+ }
+
+ public <OUT> Output<StreamRecord<OUT>>[] getBroadcastingInternalOutputs(Object output) {
+ return ReflectionUtils.getFieldValue(output, broadcastingOutputsField);
+ }
+
+ public OutputTag<?> getChainingOutputTag(Object output) {
+ return ReflectionUtils.getFieldValue(output, chainingOutputTagField);
+ }
+
+ @SuppressWarnings("unchecked")
+ public RecordWriter<SerializationDelegate<StreamElement>> getRecordWriter(Object output) {
+ return ReflectionUtils.getFieldValue(output, recordWriterField);
+ }
+
+ @SuppressWarnings("unchecked")
+ public TypeSerializer<StreamElement> getRecordWriterTypeSerializer(Object output) {
+ SerializationDelegate<StreamElement> serializationDelegate =
+ ReflectionUtils.getFieldValue(output, recordWriterSerializationDelegateField);
+ TypeSerializer<StreamElement> typeSerializer =
+ ReflectionUtils.getFieldValue(
+ serializationDelegate, serializationDelegateSerializerField);
+ return typeSerializer.duplicate();
+ }
+}
diff --git a/flink-ml-iteration/src/main/java/org/apache/flink/iteration/broadcast/RecordWriterBroadcastOutput.java b/flink-ml-iteration/src/main/java/org/apache/flink/iteration/broadcast/RecordWriterBroadcastOutput.java
new file mode 100644
index 0000000..fc99865
--- /dev/null
+++ b/flink-ml-iteration/src/main/java/org/apache/flink/iteration/broadcast/RecordWriterBroadcastOutput.java
@@ -0,0 +1,46 @@
+/*
+ * 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.iteration.broadcast;
+
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.runtime.io.network.api.writer.RecordWriter;
+import org.apache.flink.runtime.plugable.SerializationDelegate;
+import org.apache.flink.streaming.runtime.streamrecord.StreamElement;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+import java.io.IOException;
+
+/** The broadcast output corresponding to a record writer output. */
+public class RecordWriterBroadcastOutput<OUT> implements BroadcastOutput<OUT> {
+ private final RecordWriter<SerializationDelegate<StreamElement>> recordWriter;
+ private final SerializationDelegate<StreamElement> serializationDelegate;
+
+ public RecordWriterBroadcastOutput(
+ RecordWriter<SerializationDelegate<StreamElement>> recordWriter,
+ TypeSerializer<StreamElement> typeSerializer) {
+
+ this.recordWriter = recordWriter;
+ this.serializationDelegate = new SerializationDelegate<>(typeSerializer);
+ }
+
+ @Override
+ public void broadcastEmit(StreamRecord<OUT> record) throws IOException {
+ serializationDelegate.setInstance(record);
+ recordWriter.broadcastEmit(serializationDelegate);
+ }
+}
diff --git a/flink-ml-iteration/src/main/java/org/apache/flink/iteration/utils/ReflectionUtils.java b/flink-ml-iteration/src/main/java/org/apache/flink/iteration/utils/ReflectionUtils.java
new file mode 100644
index 0000000..3e61f16
--- /dev/null
+++ b/flink-ml-iteration/src/main/java/org/apache/flink/iteration/utils/ReflectionUtils.java
@@ -0,0 +1,52 @@
+/*
+ * 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.iteration.utils;
+
+import java.lang.reflect.Field;
+
+/** Utility class to provide some reflection tools. */
+public class ReflectionUtils {
+
+ public static Field getClassField(Class<?> declaredClass, String fieldName) {
+ try {
+ Field field = declaredClass.getDeclaredField(fieldName);
+ field.setAccessible(true);
+ return field;
+ } catch (Exception e) {
+ throw new RuntimeException(
+ "Failed to get field" + fieldName + " from " + declaredClass, e);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ public static <T> T getFieldValue(Object targetObject, Field field) {
+ try {
+ return (T) field.get(targetObject);
+ } catch (Exception e) {
+ throw new RuntimeException(
+ "Failed to get field " + field.getName() + " from " + targetObject, e);
+ }
+ }
+
+ public static <T> T getFieldValue(
+ Object targetObject, Class<?> declaredClass, String fieldName) {
+ Field field = getClassField(declaredClass, fieldName);
+ return getFieldValue(targetObject, field);
+ }
+}
diff --git a/flink-ml-iteration/src/test/java/org/apache/flink/iteration/broadcast/BroadcastOutputTest.java b/flink-ml-iteration/src/test/java/org/apache/flink/iteration/broadcast/BroadcastOutputTest.java
new file mode 100644
index 0000000..00a3a64
--- /dev/null
+++ b/flink-ml-iteration/src/test/java/org/apache/flink/iteration/broadcast/BroadcastOutputTest.java
@@ -0,0 +1,248 @@
+/*
+ * 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.iteration.broadcast;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.streaming.api.graph.StreamConfig;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
+import org.apache.flink.util.OutputTag;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+
+/** Tests the broadcastable operators. */
+@RunWith(Parameterized.class)
+public class BroadcastOutputTest {
+
+ private static final int NUM_RECORDS = 100;
+
+ /** Whether object reuse is used. */
+ private final boolean objectReuseEnabled;
+
+ @Parameterized.Parameters
+ public static Collection<Object[]> data() {
+ return Arrays.asList(new Object[][] {{true}, {false}});
+ }
+
+ public BroadcastOutputTest(boolean objectReuseEnabled) {
+ this.objectReuseEnabled = objectReuseEnabled;
+ }
+
+ @Test
+ public void testBroadcastWithChain() throws Exception {
+ StreamExecutionEnvironment env = createTestEnvironment();
+
+ env.addSource(new TestSource())
+ .transform(
+ "broadcast", TypeInformation.of(Integer.class), new TestBroadcastOperator())
+ .addSink(new CheckResultSink());
+ env.execute();
+ }
+
+ @Test
+ public void testBroadcastWithResultPartition() throws Exception {
+ StreamExecutionEnvironment env = createTestEnvironment();
+
+ env.addSource(new TestSource())
+ .transform(
+ "broadcast", TypeInformation.of(Integer.class), new TestBroadcastOperator())
+ .addSink(new CheckResultSink())
+ .setParallelism(2);
+ env.execute();
+ }
+
+ @Test
+ public void testBroadcastWithMultipleChain() throws Exception {
+ StreamExecutionEnvironment env = createTestEnvironment();
+
+ DataStream<Integer> dataStream =
+ env.addSource(new TestSource())
+ .transform(
+ "broadcast",
+ TypeInformation.of(Integer.class),
+ new TestBroadcastOperator());
+
+ dataStream.addSink(new CheckResultSink());
+ dataStream.addSink(new CheckResultSink());
+
+ env.execute();
+ }
+
+ @Test
+ public void testBroadcastWithMultipleResultPartitions() throws Exception {
+ StreamExecutionEnvironment env = createTestEnvironment();
+
+ DataStream<Integer> dataStream =
+ env.addSource(new TestSource())
+ .transform(
+ "broadcast",
+ TypeInformation.of(Integer.class),
+ new TestBroadcastOperator());
+
+ dataStream.addSink(new CheckResultSink()).setParallelism(2);
+ dataStream.addSink(new CheckResultSink()).setParallelism(2);
+
+ env.execute();
+ }
+
+ @Test
+ public void testBroadcastWithMixedOutput() throws Exception {
+ StreamExecutionEnvironment env = createTestEnvironment();
+
+ DataStream<Integer> dataStream =
+ env.addSource(new TestSource())
+ .transform(
+ "broadcast",
+ TypeInformation.of(Integer.class),
+ new TestBroadcastOperator());
+
+ dataStream.addSink(new CheckResultSink()).setParallelism(1);
+ dataStream.addSink(new CheckResultSink()).setParallelism(2);
+ dataStream.addSink(new CheckResultSink()).setParallelism(4);
+
+ env.execute();
+ }
+
+ @Test
+ public void testBroadcastWithMixedOutputWithSideOutput() throws Exception {
+ StreamExecutionEnvironment env = createTestEnvironment();
+
+ SingleOutputStreamOperator<Integer> dataStream =
+ env.addSource(new TestSource())
+ .transform(
+ "broadcast",
+ TypeInformation.of(Integer.class),
+ new TestBroadcastOperator());
+
+ dataStream.addSink(new CheckResultSink());
+ dataStream.getSideOutput(new OutputTag<Integer>("0") {}).addSink(new CheckResultSink());
+ dataStream
+ .getSideOutput(new OutputTag<Integer>("1") {})
+ .addSink(new CheckResultSink())
+ .setParallelism(2);
+ dataStream
+ .getSideOutput(new OutputTag<Integer>("2") {})
+ .addSink(new CheckResultSink())
+ .setParallelism(4);
+
+ env.execute();
+ }
+
+ // ------------------------------------------------------------------------
+ // Utilities
+ // ------------------------------------------------------------------------
+
+ private StreamExecutionEnvironment createTestEnvironment() {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
+
+ if (objectReuseEnabled) {
+ env.getConfig().enableObjectReuse();
+ } else {
+ env.getConfig().disableObjectReuse();
+ }
+ env.setParallelism(1);
+
+ return env;
+ }
+
+ /** A test source that emits a list of numbers. */
+ private static class TestSource implements SourceFunction<Integer> {
+ @Override
+ public void run(SourceContext<Integer> ctx) throws Exception {
+ for (int i = 0; i < NUM_RECORDS; ++i) {
+ ctx.collect(i);
+ }
+ }
+
+ @Override
+ public void cancel() {}
+ }
+
+ /** The test broadcasting operator that broadcasts all the records received. */
+ private static class TestBroadcastOperator extends AbstractStreamOperator<Integer>
+ implements OneInputStreamOperator<Integer, Integer> {
+
+ private BroadcastOutput<Integer> broadcastOutput;
+
+ @Override
+ public void setup(
+ StreamTask<?, ?> containingTask,
+ StreamConfig config,
+ Output<StreamRecord<Integer>> output) {
+ super.setup(containingTask, config, output);
+ broadcastOutput =
+ BroadcastOutputFactory.createBroadcastOutput(
+ output, metrics.getIOMetricGroup().getNumRecordsOutCounter());
+ }
+
+ @Override
+ public void processElement(StreamRecord<Integer> element) throws Exception {
+ broadcastOutput.broadcastEmit(element);
+ }
+ }
+
+ /**
+ * The test sink that checks whether all the records are received. If not, it will throw
+ * exceptions and finally cause the tests to fail.
+ */
+ private static class CheckResultSink extends RichSinkFunction<Integer> {
+ private List<Integer> received;
+
+ @Override
+ public void open(Configuration parameters) throws Exception {
+ super.open(parameters);
+ received = new ArrayList<>(NUM_RECORDS);
+ }
+
+ @Override
+ public void invoke(Integer value, Context context) {
+ received.add(value);
+ }
+
+ @Override
+ public void close() {
+ assertEquals(
+ "Number of received records does not consistent", NUM_RECORDS, received.size());
+ for (int i = 0; i < NUM_RECORDS; ++i) {
+ assertEquals(
+ String.format("The %d elements does not consistent", i),
+ received.get(i),
+ Integer.valueOf(i));
+ }
+ }
+ }
+}
diff --git a/flink-ml-iteration/src/test/java/org/apache/flink/iteration/utils/ReflectionUtilsTest.java b/flink-ml-iteration/src/test/java/org/apache/flink/iteration/utils/ReflectionUtilsTest.java
new file mode 100644
index 0000000..082ae4a
--- /dev/null
+++ b/flink-ml-iteration/src/test/java/org/apache/flink/iteration/utils/ReflectionUtilsTest.java
@@ -0,0 +1,49 @@
+/*
+ * 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.iteration.utils;
+
+import org.junit.Test;
+
+import java.lang.reflect.Field;
+
+import static org.junit.Assert.assertEquals;
+
+/** Tests the utility methods related to reflection. */
+public class ReflectionUtilsTest {
+
+ @Test
+ public void testGetField() {
+ Field field = ReflectionUtils.getClassField(Base.class, "field1");
+ assertEquals("field1", field.getName());
+ assertEquals(Base.class, field.getDeclaringClass());
+
+ Child child = new Child();
+ assertEquals("field1", ReflectionUtils.getFieldValue(child, field));
+ assertEquals("field1", ReflectionUtils.getFieldValue(child, Base.class, "field1"));
+ assertEquals("field2", ReflectionUtils.getFieldValue(child, Child.class, "field2"));
+ }
+
+ private static class Base {
+ private String field1 = "field1";
+ }
+
+ private static class Child extends Base {
+ private String field2 = "field2";
+ }
+}