You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Nil Madhab (Jira)" <ji...@apache.org> on 2025/03/10 09:09:00 UTC
[jira] [Comment Edited] (FLINK-37443) Add returns() method to DataStream V2 API for specifying output types with lambda expressions
[ https://issues.apache.org/jira/browse/FLINK-37443?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17933781#comment-17933781 ]
Nil Madhab edited comment on FLINK-37443 at 3/10/25 9:08 AM:
-------------------------------------------------------------
There is definitely some issue with lamdba expression, I was following FLIP-500: Support Join Extension in DataStream V2 API and tried to create a project with 2.1-snapshot, it works fine with anonomous class, but does not work with lambda.
Here is the [file.|https://pastes.dev/WXmhIqfh66#L193]
If I use
{code:java}
new JoinFunction<Order, ExpressEvent, EnrichedOrder>() {code}
as lambda, it breaks with no clear error message.
{code:java}
Caused by: java.lang.ClassCastException: class com.java.flinktutorials.datastreamv2.EnrichOrderExample$EnrichedOrder cannot be cast to class java.lang.Void (com.java.flinktutorials.datastreamv2.EnrichOrderExample$EnrichedOrder is in unnamed module of loader 'app'; java.lang.Void is in module java.base of loader 'bootstrap')
at org.apache.flink.api.common.typeutils.base.VoidSerializer.copy(VoidSerializer.java:30)
at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:74)
at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:50)
at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:29)
at org.apache.flink.datastream.impl.common.OutputCollector.collect(OutputCollector.java:35)
at com.java.flinktutorials.datastreamv2.EnrichOrderExample.lambda$main$5b039ba0$1(EnrichOrderExample.java:203)
at org.apache.flink.datastream.impl.extension.join.operators.TwoInputNonBroadcastJoinProcessOperator.processElement1(TwoInputNonBroadcastJoinProcessOperator.java:90)
at org.apache.flink.streaming.runtime.operators.asyncprocessing.AsyncStateProcessing.lambda$makeRecordProcessor$0(AsyncStateProcessing.java:68)
at org.apache.flink.runtime.asyncprocessing.AsyncExecutionController.lambda$syncPointRequestWithCallback$6(AsyncExecutionController.java:472)
at org.apache.flink.util.function.ThrowingConsumer.lambda$unchecked$0(ThrowingConsumer.java:54)
at org.apache.flink.core.state.StateFutureImpl.thenAccept(StateFutureImpl.java:112)
at org.apache.flink.runtime.asyncprocessing.AsyncExecutionController.syncPointRequestWithCallback(AsyncExecutionController.java:472)
at org.apache.flink.runtime.asyncprocessing.operators.AbstractAsyncStateStreamOperator.preserveRecordOrderAndProcess(AbstractAsyncStateStreamOperator.java:198)
at org.apache.flink.streaming.runtime.operators.asyncprocessing.AsyncStateProcessing.lambda$makeRecordProcessor$1(AsyncStateProcessing.java:68)
at org.apache.flink.streaming.runtime.io.StreamTwoInputProcessorFactory$StreamTaskNetworkOutput.emitRecord(StreamTwoInputProcessorFactory.java:257)
at org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.processElement(AbstractStreamTaskNetworkInput.java:206)
at org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.emitNext(AbstractStreamTaskNetworkInput.java:163)
at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65)
at org.apache.flink.streaming.runtime.io.StreamMultipleInputProcessor.processInput(StreamMultipleInputProcessor.java:86)
at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:637)
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:231)
at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:972)
at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:916)
at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:963)
at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:942)
at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:756) {code}
was (Author: nilmadhab):
There is definitely some issue with lamdba expression, I was following FLIP-500: Support Join Extension in DataStream V2 API and tried to create a project with 2.1-snapshot, it works fine with anonomous class, but does not work with lambda.
Here is the [file.|https://pastes.dev/WXmhIqfh66#L193]
If I use
new JoinFunction<Order, ExpressEvent, EnrichedOrder>() as lambda, it breaks with no clear error message.
{code:java}
Caused by: java.lang.ClassCastException: class com.java.flinktutorials.datastreamv2.EnrichOrderExample$EnrichedOrder cannot be cast to class java.lang.Void (com.java.flinktutorials.datastreamv2.EnrichOrderExample$EnrichedOrder is in unnamed module of loader 'app'; java.lang.Void is in module java.base of loader 'bootstrap')
at org.apache.flink.api.common.typeutils.base.VoidSerializer.copy(VoidSerializer.java:30)
at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:74)
at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:50)
at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:29)
at org.apache.flink.datastream.impl.common.OutputCollector.collect(OutputCollector.java:35)
at com.java.flinktutorials.datastreamv2.EnrichOrderExample.lambda$main$5b039ba0$1(EnrichOrderExample.java:203)
at org.apache.flink.datastream.impl.extension.join.operators.TwoInputNonBroadcastJoinProcessOperator.processElement1(TwoInputNonBroadcastJoinProcessOperator.java:90)
at org.apache.flink.streaming.runtime.operators.asyncprocessing.AsyncStateProcessing.lambda$makeRecordProcessor$0(AsyncStateProcessing.java:68)
at org.apache.flink.runtime.asyncprocessing.AsyncExecutionController.lambda$syncPointRequestWithCallback$6(AsyncExecutionController.java:472)
at org.apache.flink.util.function.ThrowingConsumer.lambda$unchecked$0(ThrowingConsumer.java:54)
at org.apache.flink.core.state.StateFutureImpl.thenAccept(StateFutureImpl.java:112)
at org.apache.flink.runtime.asyncprocessing.AsyncExecutionController.syncPointRequestWithCallback(AsyncExecutionController.java:472)
at org.apache.flink.runtime.asyncprocessing.operators.AbstractAsyncStateStreamOperator.preserveRecordOrderAndProcess(AbstractAsyncStateStreamOperator.java:198)
at org.apache.flink.streaming.runtime.operators.asyncprocessing.AsyncStateProcessing.lambda$makeRecordProcessor$1(AsyncStateProcessing.java:68)
at org.apache.flink.streaming.runtime.io.StreamTwoInputProcessorFactory$StreamTaskNetworkOutput.emitRecord(StreamTwoInputProcessorFactory.java:257)
at org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.processElement(AbstractStreamTaskNetworkInput.java:206)
at org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.emitNext(AbstractStreamTaskNetworkInput.java:163)
at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65)
at org.apache.flink.streaming.runtime.io.StreamMultipleInputProcessor.processInput(StreamMultipleInputProcessor.java:86)
at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:637)
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:231)
at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:972)
at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:916)
at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:963)
at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:942)
at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:756) {code}
> Add returns() method to DataStream V2 API for specifying output types with lambda expressions
> ----------------------------------------------------------------------------------------------
>
> Key: FLINK-37443
> URL: https://issues.apache.org/jira/browse/FLINK-37443
> Project: Flink
> Issue Type: Improvement
> Components: API / DataStream
> Affects Versions: 1.20.1
> Environment: * Apache Flink version: 1.20.1
> * Java version: OpenJDK 21
> * API: DataStream V2 API
> Reporter: Nil Madhab
> Priority: Major
>
> While following the official [tutorial|https://nightlies.apache.org/flink/flink-docs-master/docs/dev/datastream-v2/overview/] of DataStream V2 API, I encountered an issue.
> When using the DataStream V2 API with lambda expressions for {{{}OneInputStreamProcessFunction{}}}, Java's type erasure prevents Flink from automatically determining the output type, resulting in the following exception:
>
> {code:java}
> Exception in thread "main" org.apache.flink.api.common.functions.InvalidTypesException: The return type of function 'process(NonKeyedPartitionStreamImpl.java:74)' could not be determined automatically, due to type erasure.{code}
> {code:java}
> Caused by: org.apache.flink.api.common.functions.InvalidTypesException: The generic type parameters of 'Collector' are missing. In many cases lambda methods don't provide enough information for automatic type extraction when Java generics are involved. An easy workaround is to use an (anonymous) class instead that implements the 'org.apache.flink.datastream.api.function.OneInputStreamProcessFunction' interface. Otherwise the type has to be specified explicitly using type information.
> at org.apache.flink.api.java.typeutils.TypeExtractionUtils.validateLambdaType(TypeExtractionUtils.java:371){code}
>
>
> The error suggests implementing {{ResultTypeQueryable}} interface or using an anonymous class as workarounds. In the traditional DataStream API, users could simply call {{.returns(TypeInformation)}} to explicitly specify the output type.
> h3. Example that fails:
> {code:java}
> NonKeyedPartitionStream<Integer> parsed = input.process(
> (OneInputStreamProcessFunction<String, Integer>) (record, output, ctx) ->
> output.collect(Integer.parseInt(record))
> ); {code}
> h3.
> Example that works (but is more verbose):
> {code:java}
> NonKeyedPartitionStream<Integer> parsed = input.process(
> new OneInputStreamProcessFunction<String, Integer>() {
> @Override
> public void processRecord(String record, Collector<Integer> output,
> PartitionedContext ctx) throws Exception {
> output.collect(Integer.parseInt(record));
> }
> }
> ); {code}
> h3.
> Requested Enhancement
> Add a {{.returns(TypeInformation)}} or {{.returns(Class)}} method to the {{NonKeyedPartitionStream}} class in the DataStream V2 API to allow for the specification of output types when using lambda expressions with process functions.
> The documentation can use the anonymous class, until the issue is fixed, to prevent confusion for people new to flink.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)