[hotfix] [py] Code cleanup - PythonStreamer#sendBroadCastVariables()
Project: http://git-wip-us.apache.org/repos/asf/flink/repo Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/f4324ba7 Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/f4324ba7 Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/f4324ba7 Branch: refs/heads/table-retraction Commit: f4324ba70970ea65f305be0e793616d878319457 Parents: a7251c5 Author: zentol <ches...@apache.org> Authored: Thu Mar 30 19:38:47 2017 +0200 Committer: zentol <ches...@apache.org> Committed: Wed Apr 5 20:43:44 2017 +0200 ---------------------------------------------------------------------- .../apache/flink/python/api/streaming/data/PythonStreamer.java | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/flink/blob/f4324ba7/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/streaming/data/PythonStreamer.java ---------------------------------------------------------------------- diff --git a/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/streaming/data/PythonStreamer.java b/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/streaming/data/PythonStreamer.java index 830c843..219ae27 100644 --- a/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/streaming/data/PythonStreamer.java +++ b/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/streaming/data/PythonStreamer.java @@ -242,17 +242,17 @@ public class PythonStreamer<S extends PythonSender, OUT> implements Serializable StringSerializer stringSerializer = new StringSerializer(); for (String name : names) { - Iterator<?> bcv = function.getRuntimeContext().getBroadcastVariable(name).iterator(); + Iterator<byte[]> bcv = function.getRuntimeContext().<byte[]>getBroadcastVariable(name).iterator(); out.write(stringSerializer.serializeWithoutTypeInfo(name)); while (bcv.hasNext()) { out.writeByte(1); - out.write((byte[]) bcv.next()); + out.write(bcv.next()); } out.writeByte(0); } - } catch (SocketTimeoutException ste) { + } catch (SocketTimeoutException ignored) { throw new RuntimeException("External process for task " + function.getRuntimeContext().getTaskName() + " stopped responding." + msg); } }
