-
Notifications
You must be signed in to change notification settings - Fork 45
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Async lookups return incorrect results because JsonRowDataSerializationSchema is not thread-safe #121
Comments
Hi @grzegorz8 I've looked at KinesisFireHoseSink Connector - https://issues.apache.org/jira/browse/FLINK-24228 where it seems the It seems that My suggestion for now is to... wrap with Also maybe worth reporting this lack of functionality to Flink? |
Ok @grzegorz8 To sole this problem we have to decouple query creation and response processing from request send. I think we can extract
After that we need to change the Instead what we have now: We would need something like that:
And later modify This is a raw sketch, but I think this will solve the problem by ensuring that schema will be used only by one thread. |
I spent some time refactoring the way you suggested, but I came to the conclusion this approach requires quite huge refactoring which does not bring much benefit in comparison to calling SerializationSchema in "synchronized" way. In both approaches serialization is the bottleneck with parallelism=1. So I propose to implement the simple approach as temporary solution, and in the future try to find the way to be able to run serialization concurrently in safe way. |
I noticed that lookup in async mode may not work as expected. Namely, when multiple HTTP requests are sent at once in async mode, some of the results seems to be duplicated for other rows.
I believe the root cause is the fact that
org.apache.flink.formats.json.JsonRowDataSerializationSchema
is not thread-safe.The serialization schema is used in
com.getindata.connectors.http.internal.table.lookup.querycreators.GenericJsonQueryCreator#createLookupQuery
.The text was updated successfully, but these errors were encountered: