You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by jreiffers <gi...@git.apache.org> on 2015/06/17 15:56:50 UTC
[GitHub] flink pull request: [FLINK-2124] [streaming] Fix behavior of FromE...
GitHub user jreiffers opened a pull request:
https://github.com/apache/flink/pull/848
[FLINK-2124] [streaming] Fix behavior of FromElementsFunction when T …
…isn't Serializable
rewrite FromElementsFunction to work with arbitrary types. The elements are serialized to a byte array (using a TypeSerializer) when the FromElementsFunction is constructed and deserialized when run is called.
I also removed the (Collection<T>) constructor, because all Collections are Iterable.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/jreiffers/flink flink-2124
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/848.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #848
----
commit 171bfd115572b3d11962c28b48db3f2ceefd8280
Author: Johannes Reifferscheid <j....@gmail.com>
Date: 2015-06-17T13:35:37Z
[FLINK-2124] [streaming] Fix behavior of FromElementsFunction when T isn't Serializable
rewrite FromElementsFunction to work with arbitrary types. The elements
are serialized to a byte array (using a TypeSerializer) when the FromElementsFunction is
constructed and deserialized when run is called.
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2124] [streaming] Fix behavior of FromE...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on the pull request:
https://github.com/apache/flink/pull/848#issuecomment-113081149
LGTM. +1 for merging once Travis has given green light.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2124] [streaming] Fix behavior of FromE...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on a diff in the pull request:
https://github.com/apache/flink/pull/848#discussion_r32915549
--- Diff: flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/FromElementsFunction.java ---
@@ -17,37 +17,81 @@
package org.apache.flink.streaming.api.functions.source;
-import java.util.Arrays;
-import java.util.Collection;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.InputViewDataInputStreamWrapper;
+import org.apache.flink.core.memory.OutputViewDataOutputStreamWrapper;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.util.Iterator;
public class FromElementsFunction<T> implements SourceFunction<T> {
private static final long serialVersionUID = 1L;
- private Iterable<T> iterable;
+ private final TypeSerializer<T> serializer;
+ private final byte[] elements;
private volatile boolean isRunning = true;
- public FromElementsFunction(T... elements) {
- this.iterable = Arrays.asList(elements);
- }
+ public FromElementsFunction(TypeSerializer<T> serializer, final T... elements) {
+ this(serializer, new Iterable<T>() {
+ @Override
+ public Iterator<T> iterator() {
+ return new Iterator<T>() {
+ int index = 0;
+
+ @Override
+ public boolean hasNext() {
+ return index < elements.length;
+ }
+
+ @Override
+ public T next() {
+ return elements[index++];
+ }
- public FromElementsFunction(Collection<T> elements) {
- this.iterable = elements;
+ @Override
+ public void remove() {
+ throw new UnsupportedOperationException();
+ }
+ };
+ }
+ });
}
- public FromElementsFunction(Iterable<T> elements) {
- this.iterable = elements;
+ public FromElementsFunction(TypeSerializer<T> serializer, Iterable<T> elements) {
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ OutputViewDataOutputStreamWrapper wrapper = new OutputViewDataOutputStreamWrapper(new DataOutputStream(baos));
+
+ try {
+ for (T element : elements) {
+ serializer.serialize(element, wrapper);
+ }
+ } catch (IOException e) {
+ // ByteArrayOutputStream doesn't throw IOExceptions when written to
+ }
+ // closing the DataOutputStream would just flush the ByteArrayOutputStream, which in turn doesn't do anything.
+
+ this.serializer = serializer;
+ this.elements = baos.toByteArray();
}
@Override
public void run(SourceContext<T> ctx) throws Exception {
- Iterator<T> it = iterable.iterator();
+ T value = serializer.createInstance();
--- End diff --
Let's not use object reusing mode by default. It is not guaranteed to always work (some abstract types are not instantiable) and should only work on user-requested activation. Use `value = serializer.deserialize(input);` instead.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2124] [streaming] Fix behavior of FromE...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/flink/pull/848
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2124] [streaming] Fix behavior of FromE...
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/848#discussion_r32801136
--- Diff: flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/FromElementsFunction.java ---
@@ -17,37 +17,81 @@
package org.apache.flink.streaming.api.functions.source;
-import java.util.Arrays;
-import java.util.Collection;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.InputViewDataInputStreamWrapper;
+import org.apache.flink.core.memory.OutputViewDataOutputStreamWrapper;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.util.Iterator;
public class FromElementsFunction<T> implements SourceFunction<T> {
private static final long serialVersionUID = 1L;
- private Iterable<T> iterable;
+ private final TypeSerializer<T> serializer;
+ private final byte[] elements;
private volatile boolean isRunning = true;
- public FromElementsFunction(T... elements) {
- this.iterable = Arrays.asList(elements);
- }
+ public FromElementsFunction(TypeSerializer<T> serializer, final T... elements) {
+ this(serializer, new Iterable<T>() {
+ @Override
+ public Iterator<T> iterator() {
+ return new Iterator<T>() {
+ int index = 0;
+
+ @Override
+ public boolean hasNext() {
+ return index < elements.length;
+ }
+
+ @Override
+ public T next() {
+ return elements[index++];
+ }
- public FromElementsFunction(Collection<T> elements) {
- this.iterable = elements;
+ @Override
+ public void remove() {
+ throw new UnsupportedOperationException();
+ }
+ };
+ }
+ });
}
- public FromElementsFunction(Iterable<T> elements) {
- this.iterable = elements;
+ public FromElementsFunction(TypeSerializer<T> serializer, Iterable<T> elements) {
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ OutputViewDataOutputStreamWrapper wrapper = new OutputViewDataOutputStreamWrapper(new DataOutputStream(baos));
+
+ try {
+ for (T element : elements) {
+ serializer.serialize(element, wrapper);
+ }
+ } catch (IOException e) {
+ // ByteArrayOutputStream doesn't throw IOExceptions when written to
--- End diff --
I think not forwarding the exception here is a bad idea.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2124] [streaming] Fix behavior of FromE...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on a diff in the pull request:
https://github.com/apache/flink/pull/848#discussion_r32915439
--- Diff: flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/FromElementsFunction.java ---
@@ -17,37 +17,81 @@
package org.apache.flink.streaming.api.functions.source;
-import java.util.Arrays;
-import java.util.Collection;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.InputViewDataInputStreamWrapper;
+import org.apache.flink.core.memory.OutputViewDataOutputStreamWrapper;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.util.Iterator;
public class FromElementsFunction<T> implements SourceFunction<T> {
private static final long serialVersionUID = 1L;
- private Iterable<T> iterable;
+ private final TypeSerializer<T> serializer;
+ private final byte[] elements;
private volatile boolean isRunning = true;
- public FromElementsFunction(T... elements) {
- this.iterable = Arrays.asList(elements);
- }
+ public FromElementsFunction(TypeSerializer<T> serializer, final T... elements) {
+ this(serializer, new Iterable<T>() {
--- End diff --
I think you can make this much simpler by using `Arrays.asList(elements)`
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2124] [streaming] Fix behavior of FromE...
Posted by jreiffers <gi...@git.apache.org>.
Github user jreiffers commented on a diff in the pull request:
https://github.com/apache/flink/pull/848#discussion_r32916320
--- Diff: flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/FromElementsFunction.java ---
@@ -17,37 +17,81 @@
package org.apache.flink.streaming.api.functions.source;
-import java.util.Arrays;
-import java.util.Collection;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.InputViewDataInputStreamWrapper;
+import org.apache.flink.core.memory.OutputViewDataOutputStreamWrapper;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.util.Iterator;
public class FromElementsFunction<T> implements SourceFunction<T> {
private static final long serialVersionUID = 1L;
- private Iterable<T> iterable;
+ private final TypeSerializer<T> serializer;
+ private final byte[] elements;
private volatile boolean isRunning = true;
- public FromElementsFunction(T... elements) {
- this.iterable = Arrays.asList(elements);
- }
+ public FromElementsFunction(TypeSerializer<T> serializer, final T... elements) {
+ this(serializer, new Iterable<T>() {
+ @Override
+ public Iterator<T> iterator() {
+ return new Iterator<T>() {
+ int index = 0;
+
+ @Override
+ public boolean hasNext() {
+ return index < elements.length;
+ }
+
+ @Override
+ public T next() {
+ return elements[index++];
+ }
- public FromElementsFunction(Collection<T> elements) {
- this.iterable = elements;
+ @Override
+ public void remove() {
+ throw new UnsupportedOperationException();
+ }
+ };
+ }
+ });
}
- public FromElementsFunction(Iterable<T> elements) {
- this.iterable = elements;
+ public FromElementsFunction(TypeSerializer<T> serializer, Iterable<T> elements) {
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ OutputViewDataOutputStreamWrapper wrapper = new OutputViewDataOutputStreamWrapper(new DataOutputStream(baos));
+
+ try {
+ for (T element : elements) {
+ serializer.serialize(element, wrapper);
+ }
+ } catch (IOException e) {
+ // ByteArrayOutputStream doesn't throw IOExceptions when written to
--- End diff --
Right, I missed that.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2124] [streaming] Fix behavior of FromE...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/848#discussion_r32709026
--- Diff: flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/FromElementsFunction.java ---
@@ -17,37 +17,71 @@
package org.apache.flink.streaming.api.functions.source;
-import java.util.Arrays;
-import java.util.Collection;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.InputViewDataInputStreamWrapper;
+import org.apache.flink.core.memory.OutputViewDataOutputStreamWrapper;
+
+import java.io.*;
import java.util.Iterator;
public class FromElementsFunction<T> implements SourceFunction<T> {
private static final long serialVersionUID = 1L;
- private Iterable<T> iterable;
+ private final TypeSerializer<T> serializer;
+ private final byte[] elements;
private volatile boolean isRunning = true;
- public FromElementsFunction(T... elements) {
- this.iterable = Arrays.asList(elements);
- }
+ public FromElementsFunction(TypeSerializer<T> serializer, final T... elements) {
+ this(serializer, new Iterable<T>() {
+ @Override
+ public Iterator<T> iterator() {
+ return new Iterator<T>() {
+ int index = 0;
+
+ @Override
+ public boolean hasNext() {
+ return index < elements.length;
+ }
- public FromElementsFunction(Collection<T> elements) {
- this.iterable = elements;
+ @Override
+ public T next() {
+ return elements[index++];
+ }
+ };
+ }
+ });
}
- public FromElementsFunction(Iterable<T> elements) {
- this.iterable = elements;
+ public FromElementsFunction(TypeSerializer<T> serializer, Iterable<T> elements) {
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ OutputViewDataOutputStreamWrapper wrapper = new OutputViewDataOutputStreamWrapper(new DataOutputStream(baos));
+
+ try {
+ for (T element : elements)
+ serializer.serialize(element, wrapper);
--- End diff --
Implicit coding conventions are to put curly brackets also around one line `for` statements.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2124] [streaming] Fix behavior of FromE...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on the pull request:
https://github.com/apache/flink/pull/848#issuecomment-113079060
Thanks for your contribution @jreiffers. The PR looks really good :-) There is only a problem with the `.*` import.
I was thinking about adding a test case where we create a `FromElementsFunction` with non serializable data elements. What do you think?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2124] [streaming] Fix behavior of FromE...
Posted by mbalassi <gi...@git.apache.org>.
Github user mbalassi commented on a diff in the pull request:
https://github.com/apache/flink/pull/848#discussion_r32817766
--- Diff: flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/FromElementsFunction.java ---
@@ -17,37 +17,81 @@
package org.apache.flink.streaming.api.functions.source;
-import java.util.Arrays;
-import java.util.Collection;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.InputViewDataInputStreamWrapper;
+import org.apache.flink.core.memory.OutputViewDataOutputStreamWrapper;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.util.Iterator;
public class FromElementsFunction<T> implements SourceFunction<T> {
private static final long serialVersionUID = 1L;
- private Iterable<T> iterable;
+ private final TypeSerializer<T> serializer;
+ private final byte[] elements;
private volatile boolean isRunning = true;
- public FromElementsFunction(T... elements) {
- this.iterable = Arrays.asList(elements);
- }
+ public FromElementsFunction(TypeSerializer<T> serializer, final T... elements) {
+ this(serializer, new Iterable<T>() {
+ @Override
+ public Iterator<T> iterator() {
+ return new Iterator<T>() {
+ int index = 0;
+
+ @Override
+ public boolean hasNext() {
+ return index < elements.length;
+ }
+
+ @Override
+ public T next() {
+ return elements[index++];
+ }
- public FromElementsFunction(Collection<T> elements) {
- this.iterable = elements;
+ @Override
+ public void remove() {
+ throw new UnsupportedOperationException();
--- End diff --
Please add some explanation message to the exception. On paper it is for intrnal use, but still. :)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2124] [streaming] Fix behavior of FromE...
Posted by jreiffers <gi...@git.apache.org>.
Github user jreiffers commented on a diff in the pull request:
https://github.com/apache/flink/pull/848#discussion_r32916185
--- Diff: flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/FromElementsFunction.java ---
@@ -17,37 +17,81 @@
package org.apache.flink.streaming.api.functions.source;
-import java.util.Arrays;
-import java.util.Collection;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.InputViewDataInputStreamWrapper;
+import org.apache.flink.core.memory.OutputViewDataOutputStreamWrapper;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.util.Iterator;
public class FromElementsFunction<T> implements SourceFunction<T> {
private static final long serialVersionUID = 1L;
- private Iterable<T> iterable;
+ private final TypeSerializer<T> serializer;
+ private final byte[] elements;
private volatile boolean isRunning = true;
- public FromElementsFunction(T... elements) {
- this.iterable = Arrays.asList(elements);
- }
+ public FromElementsFunction(TypeSerializer<T> serializer, final T... elements) {
+ this(serializer, new Iterable<T>() {
--- End diff --
Maybe. Arrays.asList also unnecessarily copies the array.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2124] [streaming] Fix behavior of FromE...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/848#discussion_r32708806
--- Diff: flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/FromElementsFunction.java ---
@@ -17,37 +17,71 @@
package org.apache.flink.streaming.api.functions.source;
-import java.util.Arrays;
-import java.util.Collection;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.InputViewDataInputStreamWrapper;
+import org.apache.flink.core.memory.OutputViewDataOutputStreamWrapper;
+
+import java.io.*;
import java.util.Iterator;
public class FromElementsFunction<T> implements SourceFunction<T> {
private static final long serialVersionUID = 1L;
- private Iterable<T> iterable;
+ private final TypeSerializer<T> serializer;
+ private final byte[] elements;
private volatile boolean isRunning = true;
- public FromElementsFunction(T... elements) {
- this.iterable = Arrays.asList(elements);
- }
+ public FromElementsFunction(TypeSerializer<T> serializer, final T... elements) {
+ this(serializer, new Iterable<T>() {
+ @Override
+ public Iterator<T> iterator() {
+ return new Iterator<T>() {
+ int index = 0;
+
+ @Override
+ public boolean hasNext() {
+ return index < elements.length;
+ }
- public FromElementsFunction(Collection<T> elements) {
- this.iterable = elements;
+ @Override
+ public T next() {
+ return elements[index++];
+ }
+ };
+ }
+ });
}
- public FromElementsFunction(Iterable<T> elements) {
- this.iterable = elements;
+ public FromElementsFunction(TypeSerializer<T> serializer, Iterable<T> elements) {
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ OutputViewDataOutputStreamWrapper wrapper = new OutputViewDataOutputStreamWrapper(new DataOutputStream(baos));
+
+ try {
+ for (T element : elements)
+ serializer.serialize(element, wrapper);
+ } catch (IOException e) {
+ // ByteArrayOutputStream doesn't throw IOExceptions when written to
+ }
+ // closing the DataOutputStream would just flush the ByteArrayOutputStream, which in turn doesn't do anything.
+
+ this.serializer = serializer;
+ this.elements = baos.toByteArray();
}
@Override
public void run(SourceContext<T> ctx) throws Exception {
- Iterator<T> it = iterable.iterator();
+ T value = serializer.createInstance();
+ ByteArrayInputStream bais = new ByteArrayInputStream(elements);
+ DataInputView input = new InputViewDataInputStreamWrapper(new DataInputStream(bais));
- while (isRunning && it.hasNext()) {
- ctx.collect(it.next());
+ while (isRunning && bais.available() > 0) {
+ value = serializer.deserialize(value, input);
+ ctx.collect(value);
}
+ // closing the DataOutputStream would just close the ByteArrayInputStream, which doesn't do anything
--- End diff --
I guess you meant the `DataInputStream`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2124] [streaming] Fix behavior of FromE...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/848#discussion_r32709133
--- Diff: flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/FromElementsFunction.java ---
@@ -17,37 +17,71 @@
package org.apache.flink.streaming.api.functions.source;
-import java.util.Arrays;
-import java.util.Collection;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.InputViewDataInputStreamWrapper;
+import org.apache.flink.core.memory.OutputViewDataOutputStreamWrapper;
+
+import java.io.*;
--- End diff --
The checkstyle plugin complains about the `.*` import here. You have to import the different streams explicitly.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2124] [streaming] Fix behavior of FromE...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on the pull request:
https://github.com/apache/flink/pull/848#issuecomment-114044907
I think a good addition would be to remember the number of elements in the set, and use that to validate that the deserialization worked properly. Serializers may be user-defined (when they want) and the user-defined serializers may be buggy. Counting the number of elements and verifying that everything is consumed after deserialization is a good way of figuring out whether things are sane, and giving an error if not.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2124] [streaming] Fix behavior of FromE...
Posted by jreiffers <gi...@git.apache.org>.
Github user jreiffers commented on a diff in the pull request:
https://github.com/apache/flink/pull/848#discussion_r32916595
--- Diff: flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/FromElementsFunction.java ---
@@ -17,37 +17,81 @@
package org.apache.flink.streaming.api.functions.source;
-import java.util.Arrays;
-import java.util.Collection;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.InputViewDataInputStreamWrapper;
+import org.apache.flink.core.memory.OutputViewDataOutputStreamWrapper;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.util.Iterator;
public class FromElementsFunction<T> implements SourceFunction<T> {
private static final long serialVersionUID = 1L;
- private Iterable<T> iterable;
+ private final TypeSerializer<T> serializer;
+ private final byte[] elements;
private volatile boolean isRunning = true;
- public FromElementsFunction(T... elements) {
- this.iterable = Arrays.asList(elements);
- }
+ public FromElementsFunction(TypeSerializer<T> serializer, final T... elements) {
+ this(serializer, new Iterable<T>() {
--- End diff --
As it turns out, it actually doesn't... the more you know!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---