[
https://issues.apache.org/jira/browse/FLINK-18770?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17167889#comment-17167889
]
Leonid Ilyevsky commented on FLINK-18770:
-----------------------------------------
Please see the main function of my app in AppMain.java. I guess, this is the
only relevant part for the understanding of how I run it.
In the beginning you will see
{quote}val env = StreamExecutionEnvironment.getExecutionEnvironment
env.enableCheckpointing(10000, CheckpointingMode.EXACTLY_ONCE)
env.setParallelism(config.bridgeConfig.parallelism)
{quote}
then it creates the pipeline, and in the end:
{quote} val _ = env.execute(config.bridgeConfig.instanceId)
{quote}
In the local environment I simply run this code inside IntelliJ.
To run it under cluster, I use python script that generates the command, you
can see the actual command in run_command.txt file, as printed by the script
before launching it.
So I basically use org.apache.flink.client.cli.CliFrontend to submit the job.
> Emitting element fails in KryoSerializer
> ----------------------------------------
>
> Key: FLINK-18770
> URL: https://issues.apache.org/jira/browse/FLINK-18770
> Project: Flink
> Issue Type: Bug
> Components: API / Type Serialization System
> Affects Versions: 1.11.1
> Environment: Flink 1.11.1, Linux
> Reporter: Leonid Ilyevsky
> Priority: Major
> Attachments: AppMain.java, KryoException.txt, SolaceSource.java,
> run_command.txt
>
>
> I wrote a simple Flink connector for Solace, see attached java file. It works
> fine under local execution environment. However, when I deployed it in the
> real Flink cluster, it failed with the Kryo exception, see attached.
> After a few hours of search and debugging, I can see now what is going on.
> The data I want to emit from this source is a simple byte array. In the
> exception stack you can see that when I call 'collect' on the context, it
> goes into OperatorChain.java:715, and then to KryoSerializer, where it
> ultimately fails. I didn't have a chance to learn what KryoSerializer is and
> why it would not know what to do with byte[], but that is not the point now.
> Then I used debugger in my local test, in order to figure out how it manages
> to work. I saw that after OperatorChain.java:715 it goes into
> BytePrimitiveArraySerializer, and then everything is working as expected.
> Obviously BytePrimitiveArraySerializer makes sense for byte[] data.
> The question is, how can I configure the execution environment under cluster
> so that it does serialization the same way as the local one? I looked at
> [https://ci.apache.org/projects/flink/flink-docs-stable/dev/execution_configuration.html]
> , and I was thinking of setting disableForceKryo, but it says it is disabled
> by default anyway.
>
> Another question is, why cluster execution environment has different default
> settings compare to local? This makes it difficult to rely on local tests.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
