becketqin commented on issue #8331: [FLINK-11987][flink-connector-kafka-0.11] convert NEXT_TRANSACTIONAL_ID_HINT_DESCRIPTOR to a non-static variabl… URL: https://github.com/apache/flink/pull/8331#issuecomment-496821999 @Zor-X-L Thanks for the patch. The issue is essentially `StateDescriptor` is mutable and not thread safe. The patch fixes the issue in `FlinkKafkaProducer011`. There might be somewhere else in the code having the same issue. I agree with @StephanEwen that a better fix would be in `StateDescriptor` itself. There are two possible fixes: Option 1: Ideally, `StateDescriptor` should be immutable. The `typeInfo` field is only used to create `TypeSerializer`. We can probably do this in the constructor instead of saving the `typeInfo` and creating the serializer lazily. However, because creating `TypeSerializer` from `TypeInfo` depends on the execution config which is usually not available when the `StateDescriptor` is constructed, the TypeSerializer have to be created lazily at runtime, unless we make some larger interface change. Option 2: Fix `StateDescriptor` to make it thread safe. This is more feasible at this point. There are currently three fields that might be modified after construction. `ttlConfig`, `querableName` and `typeSerializer`. I think we can just synchronize on them to make sure they are only set once. It still has caveats that users may set multiple different values while only the first value will take effect, but it is no worse than the current situation. @StephanEwen what do you think?
---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: [email protected] With regards, Apache Git Services
