You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@flink.apache.org by Alexey Trenikhun <ye...@msn.com> on 2019/02/02 18:39:23 UTC

Test harness for CoProcessFunction outputting Protobuf messages

Hello,
I have CoProcessFunction which outputs protobuf messages, and custom TypeInformation implementation to serialize protobuf messages, when I define graph I can set output type using SingleOutputStreamOperator.returns(TypeInformation<T>), for side outputs I can also specify TypeInformation<T> in constructor of OutputTag. However in test when I'm trying to use KeyedTwoInputStreamOperatorTestHarness I can't find way to set output type via explicit TypeInformation, furthermore if I'm trying to use side output, public <X> void AbstractStreamOperatorTestHarness::collect(OutputTag<X> outputTag, StreamRecord<X> record) seems ignore type information in outputTag and instead tries to extract type information from value using TypeExtractor.getForObject(record.getValue()). Is it expected behavior? Is another  way specify type information for test harness?

Thanks,
Alexey

Re: Test harness for CoProcessFunction outputting Protobuf messages

Posted by Alexey Trenikhun <ye...@msn.com>.
Sure - https://jira.apache.org/jira/browse/FLINK-11523

Thanks,Alexey

________________________________
From: Fabian Hueske <fh...@gmail.com>
Sent: Monday, February 4, 2019 5:59 AM
To: Alexey Trenikhun
Cc: user@flink.apache.org
Subject: Re: Test harness for CoProcessFunction outputting Protobuf messages

Hi Alexey,

I think you are right. It does not seem to be possible to provide a TypeInformation for side outputs to a TestHarness.

This sounds like a useful addition.
Would you mind creating a Jira issue for that?

Thank you,
Fabian

Am So., 3. Feb. 2019 um 19:13 Uhr schrieb Alexey Trenikhun <ye...@msn.com>>:
TestHarness has setup(TypeSerializer outputSerializer) method, which allows to set serializer for primary output. Still unclear how to set side output serializers for MockOutput.

Alexey


________________________________
From: Alexey Trenikhun <ye...@msn.com>>
Sent: Saturday, February 2, 2019 10:39 AM
To: user@flink.apache.org<ma...@flink.apache.org>
Subject: Test harness for CoProcessFunction outputting Protobuf messages

Hello,
I have CoProcessFunction which outputs protobuf messages, and custom TypeInformation implementation to serialize protobuf messages, when I define graph I can set output type using SingleOutputStreamOperator.returns(TypeInformation<T>), for side outputs I can also specify TypeInformation<T> in constructor of OutputTag. However in test when I'm trying to use KeyedTwoInputStreamOperatorTestHarness I can't find way to set output type via explicit TypeInformation, furthermore if I'm trying to use side output, public <X> void AbstractStreamOperatorTestHarness::collect(OutputTag<X> outputTag, StreamRecord<X> record) seems ignore type information in outputTag and instead tries to extract type information from value using TypeExtractor.getForObject(record.getValue()). Is it expected behavior? Is another  way specify type information for test harness?

Thanks,
Alexey

Re: Test harness for CoProcessFunction outputting Protobuf messages

Posted by Fabian Hueske <fh...@gmail.com>.
Hi Alexey,

I think you are right. It does not seem to be possible to provide a
TypeInformation for side outputs to a TestHarness.

This sounds like a useful addition.
Would you mind creating a Jira issue for that?

Thank you,
Fabian

Am So., 3. Feb. 2019 um 19:13 Uhr schrieb Alexey Trenikhun <ye...@msn.com>:

> TestHarness has setup(TypeSerializer outputSerializer) method, which
> allows to set serializer for primary output. Still unclear how to set side
> output serializers for MockOutput.
>
> Alexey
>
>
> ------------------------------
> *From:* Alexey Trenikhun <ye...@msn.com>
> *Sent:* Saturday, February 2, 2019 10:39 AM
> *To:* user@flink.apache.org
> *Subject:* Test harness for CoProcessFunction outputting Protobuf
> messages
>
> Hello,
> I have CoProcessFunction which outputs protobuf messages, and custom
> TypeInformation implementation to serialize protobuf messages, when I
> define graph I can set output type
> using SingleOutputStreamOperator.returns(TypeInformation<T>), for side
> outputs I can also specify TypeInformation<T> in constructor of OutputTag.
> However in test when I'm trying to
> use KeyedTwoInputStreamOperatorTestHarness I can't find way to set output
> type via explicit TypeInformation, furthermore if I'm trying to use side
> output, *public <X> void
> AbstractStreamOperatorTestHarness::collect(OutputTag<X> outputTag,
> StreamRecord<X> record)* seems ignore type information in outputTag and
> instead tries to extract type information from value using
> *TypeExtractor.getForObject(record.getValue())*. Is it expected behavior?
> Is another  way specify type information for test harness?
>
> Thanks,
> Alexey
>

Re: Test harness for CoProcessFunction outputting Protobuf messages

Posted by Alexey Trenikhun <ye...@msn.com>.
TestHarness has setup(TypeSerializer outputSerializer) method, which allows to set serializer for primary output. Still unclear how to set side output serializers for MockOutput.

Alexey


________________________________
From: Alexey Trenikhun <ye...@msn.com>
Sent: Saturday, February 2, 2019 10:39 AM
To: user@flink.apache.org
Subject: Test harness for CoProcessFunction outputting Protobuf messages

Hello,
I have CoProcessFunction which outputs protobuf messages, and custom TypeInformation implementation to serialize protobuf messages, when I define graph I can set output type using SingleOutputStreamOperator.returns(TypeInformation<T>), for side outputs I can also specify TypeInformation<T> in constructor of OutputTag. However in test when I'm trying to use KeyedTwoInputStreamOperatorTestHarness I can't find way to set output type via explicit TypeInformation, furthermore if I'm trying to use side output, public <X> void AbstractStreamOperatorTestHarness::collect(OutputTag<X> outputTag, StreamRecord<X> record) seems ignore type information in outputTag and instead tries to extract type information from value using TypeExtractor.getForObject(record.getValue()). Is it expected behavior? Is another  way specify type information for test harness?

Thanks,
Alexey