[Feature][Format] Improve maxwell_json,canal_json,debezium_json format support merge update_before and update_after #9805
Conversation
| - [x] canal_json | ||
| - [x] debezium_json | ||
| - [x] maxwell_json |
There was a problem hiding this comment.
Have you checked the behavior of canal_json and maxwell_json? I'm not sure about it should merge or not.
There was a problem hiding this comment.
Have you checked the behavior of
canal_jsonandmaxwell_json? I'm not sure about it should merge or not.
I don't understand what you mean. merge only add update event data, canal_json and maxwell_json support update type
There was a problem hiding this comment.
Never mind, I checked the canal_json and maxwell_json support UPDATE (without UPDATE_BEFORE and UPDATE_AFTER).
Hisoka-X
changed the title
| this.options = new HashMap<>(); | ||
| this.options.put( | ||
| CanalJsonFormatOptions.MERGE_UPDATE_EVENT.key(), | ||
| textFileSinkConfig.getMergeUpdateEvent().toString()); |
There was a problem hiding this comment.
why introduce options? Let's keep the code simple.
There was a problem hiding this comment.
why introduce options? Let's keep the code simple.
if not, the CanalJsonSerializationSchema() will add a param. if we want change other , will add a param,and the CanalJsonSerializationSchema() will have many param. use Map can keep the code simple.
There was a problem hiding this comment.
Let's add it when we really need it. But I think we don't need this for now.
There was a problem hiding this comment.
Let's add it when we really need it. But I think we don't need this for now.
i fixed it
| @@ -158,6 +157,7 @@ private void parsePayload(Collector<SeaTunnelRow> out, TablePath tablePath, Json | |||
| if (tsNode != null) { | |||
| MetadataUtil.setEventTime(after, tsNode.asLong()); | |||
| } | |||
| out.collect(before); | |||
There was a problem hiding this comment.
Why moved this?
if parse DATA_AFTER have error, out.collect(before) and out.collect(after) together can make sure before and after data send to target , not only send before
| @@ -204,6 +203,7 @@ public void deserializeMessage( | |||
| if (tsNode != null) { | |||
| MetadataUtil.setEventTime(after, ts); | |||
| } | |||
| out.collect(before); | |||
Purpose of this pull request
close #9773
Does this PR introduce any user-facing change?
How was this patch tested?
Check list
New License Guide