[Bug][mongo] Fix MongoDB's rewriteOutputBuffer to emit +I rather than +U(or -U)

[loserwang1024]

Fix #2759

[loserwang1024]

@Jiabao-Sun , @Shawn-Hx , CC

[Jiabao-Sun]

+U and +I should have the same processing in the external system.
Please give an example to explain the specific impact?

[loserwang1024]

+U and +I should have the same processing in the external system

@Jiabao-Sun , this +U is without same key data(+I) before, which is not required the changelog semantic. What happens if the downstream need the data before when process +U record?

I wonder whether a +U (without +I before) will cause other problems in streaming compute or not? @leonardBang , CC , WDYT?

[loserwang1024]

@Jiabao-Sun , moreover, if enableFullDocPrePostImage = true, also will use MongoDBConnectorFullChangelogDeserializationSchema and emit -U without +I before.

[Jiabao-Sun]

@Jiabao-Sun , moreover, if enableFullDocPrePostImage = true, also will use MongoDBConnectorFullChangelogDeserializationSchema and emit -U without +I before.

-U and +U contains the same key, it should be covered by +U ?

[loserwang1024]

-U and +U contains the same key, it should be covered by +U ?

@Jiabao-Sun , MongoDBFetchTaskContext#rewriteOutputBuffer method just covered with modify opration(it may contains +U or -U&+U, depending on config "scan.full-changelog"). Then MongoDBConnectorDeserializationSchema#deserialize will only get +U from it , while MongoDBConnectorFullChangelogDeserializationSchema#deserialize will get +U and -U from it

[Shawn-Hx]

I agree with @loserwang1024. Logically, data before highwatermark are snapshot data, which should be seen as +I. Otherwise the downstream operators may see a +U/-U record without +I before. I'm not sure whether this behavior may cause data incorrectness or not, but using +I is correct with no doubt. @Jiabao-Sun

[Jiabao-Sun]

OK, @loserwang1024 could you help add some tests for this?

[loserwang1024]

@Jiabao-Sun , thanks a lot. It's not easy to add test now.
In #2553, I add hooks to do database operations during snapshot split code and also add test in it. However, that PR depends on this PR.Therefore, please merge this PR at first.
The detail can see this branch: https://github.com/loserwang1024/flink-cdc-connectors/tree/add-snapshot-hooks. MongoDBFullChangelogITCase#testEnableBackfillWithPreHighWaterMark、testEnableBackfillWithPostLowWaterMark is what you need.

[Jiabao-Sun]

Suggested change

	value.put(MongoDBEnvelope.OPERATION_TYPE_FIELD, INSERT.getValue());
	value.put(MongoDBEnvelope.OPERATION_TYPE_FIELD, OperationType.INSERT.getValue());

[loserwang1024]

Thanks a lot , I will do it