Github user mxm commented on the pull request:
https://github.com/apache/flink/pull/1913#issuecomment-213003239
The problem is a bit more involved. We have basically three possible
branches for the `ExecutionConfig` usage.
1. Serialization during `JobGraph`/`StreamGraph` generation and
deserialization using the user code class loader during instantiating of the
tasks
2. Usage in `PojoSerializer` where no explicit
serialization/deserialization is performed because it is assumed that the
correct class loader is in place.
3. Reuse of the `ExecutionConfig` for further jobs
If we alter the `ExecutionConfig` after 1) by setting the fields to `null`,
we change the configuration for the next job. The `ExecutionEnvironment` reuses
the config. This problem is not always visible because it depends on Akka
whether the class is serialized or simply passed as a reference. If the class
is serialized, then a deserialization of the lists won't affect the original
reference.
As a solution, I've wrapped the types/serializer lists in a
`SerilizableCacheableValue` which keeps the value for as long as possible and
deserializes using the default class loader when not explicitly deserialized
during task instantiating.
---
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 [email protected] or file a JIRA ticket
with INFRA.
---
