[FLINK-21386][datastream] Postpone FromElementsFunction serialization to respect later type customization #15507
Conversation
…ects OutputTypeConfigurable to function
… to respect later type customization This fixes cases where earlier determined serializer was incorrect and returns() was ignored.
|
Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community Automated ChecksLast check on commit 493e760 (Fri May 28 09:00:27 UTC 2021) Warnings:
Mention the bot in a comment to re-run the automated checks. Review Progress
Please see the Pull Request Review Guide for a full explanation of the review process. The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required Bot commandsThe @flinkbot bot supports the following commands:
|
| } | ||
|
|
||
| @Test | ||
| public void testSetOutputTypeWithExistingBrokenSerializer() throws Exception { |
There was a problem hiding this comment.
How is this test different from the one above? In the context of setOutputType?
Could we have one test (instead of testSetOutputTypeWithExistingBrokenSerializer & testSetOutputTypeWithDifferentSerializer) that verifies that the updated serializer is used? E.g. a custom string serializer that adds additional characters, or something similar?
There was a problem hiding this comment.
testSetOutputTypeWithExistingBrokenSerializer will breaks if there is no setOutputType after while testSetOutputTypeWithDifferentSerializer will succeed(assumed that there is not assertion for different serializer).
testSetOutputTypeWithExistingBrokenSerializer matches scenario @zentol posted in FLINK-21386 that eagerly determined is incorrect. I reuse DeserializeTooMuchType for its incorrectness(see testDeSerializationError).
There are do duplication between the two. How about removing testSetOutputTypeWithDifferentSerializer and adding assertion for different serializer in testSetOutputTypeWithExistingBrokenSerializer ?
There was a problem hiding this comment.
Merged to one testSetOutputTypeWithExistingBrokenSerializer.
...ng-java/src/test/java/org/apache/flink/streaming/api/functions/FromElementsFunctionTest.java
Outdated
Show resolved
Hide resolved
...ng-java/src/test/java/org/apache/flink/streaming/api/functions/FromElementsFunctionTest.java
Outdated
Show resolved
Hide resolved
| @@ -209,6 +211,25 @@ public void testVirtualTransformations() throws Exception { | |||
| instanceof ShufflePartitioner); | |||
| } | |||
|
|
|||
| @Test | |||
| public void testOutputTypeConfigurationWithUdfStreamOperator() throws Exception { | |||
There was a problem hiding this comment.
How is this test related to the fromElements function?
There was a problem hiding this comment.
FromElementsFunction relies on feature that AbstractUdfStreamOperator will redirect OutputTypeConfigurable to its function. But StreamGraphGeneratorTest does not have test for this while it has tests for StreamOperator(testOutputTypeConfigurationWithOneInputTransformation and testOutputTypeConfigurationWithTwoInputTransformation). This test resides in a separate commit.
| } catch (IOException e) { | ||
| throw new RuntimeException(e.getMessage(), e); | ||
| } | ||
| SourceFunction<OUT> function = new FromElementsFunction<>(data); |
There was a problem hiding this comment.
Could we add a single positive test for the returns method on fromElements?
There was a problem hiding this comment.
Add StreamExecutionEnvironmentTest.testFromElementsPostConstructionType.
…r redirects OutputTypeConfigurable to function
…ization to respect later type customization Changes after reviewing: * Add `testFromElementsDeducedType` and `testFromElementsPostConstructionType` in `StreamExecutionEnvironmentTest`. * More meaningful exception message for calling `setOutputType` after serialization/deserialization. * Use `junit.rules.ExpectedException` instead of try-catch. * Merge `testSetOutputTypeWithExistingBrokenSerializer` and `testSetOutputTypeWithDifferentSerializer` to one.
|
Hi @dawidwys, I have pushed fixup commit for next reviewing cycle. Please confirm whether I have resolve your concerns. Thanks. |
|
Test was cancelled in kafka_gelly. |
What is the purpose of the change
Respect type customization(eg.
SingleOutputStreamOperator.returns) forFromElementsFunction.Brief change log
AbstractUdfStreamOperatorredirectsOutputTypeConfigurableto functionFromElementsFunctionserialization to respect later type customizationVerifying this change
This change added tests and can be verified as follows:
StreamGraphGeneratorTesttestOutputTypeConfigurationWithUdfStreamOperatorFromElementsFunctionTestFromElementsFunctiondirectly or indirectly.Does this pull request potentially affect one of the following parts:
@Public(Evolving): (yes)Documentation
WARNING: possible breaking change
Previously,
FromElementsFunctionserializes elements eagerly. After this pr, it serializes elements lazily. Any code which depend on previous behavior may break. Depends on perspective, this might be treated as a breaking change. Following are possible affected apis:StreamExecutionEnvironment.fromElements(OUT... data)StreamExecutionEnvironment.fromElements(Class<OUT> type, OUT... data)StreamExecutionEnvironment.fromCollection(Collection<OUT> data)StreamExecutionEnvironment.fromCollection(Collection<OUT> data, TypeInformation<OUT> typeInfo)