Kafka sink expected format

[Ilanlido]

Hi, i'm trying to sync MongoDB -> clickhouse and this sink connector looked promising but so far didn't manage to get it working..
i've compiled the jars myself and added it to my kafka-connect plugins folder, after adding some debug messages i found out that ClickHouseConverter returns null on all of my messages, it looks like it expects the schema to be

before={},
after={},
op=u 

however, mongo debezium source connector writes the before and after as plain strings and not objects - also in the schema its string, it has this transformer ExtractNewDocumentState that does actually deserialize the after object, but then message schema is straight the object with the field __op, which then again is not the expected format of this sink connector (doesn't include "before", "after", "op" fields)
How can i overcome this issue? should i patch these specific cases for my use or is there a better way to deal with it?

Thanks

[subkanthi]

Hi @Ilanlido , we really havent focused on Mongo support with the kafka version, in the lightweight jar, for the clickHouseConverter, it should be the schema as you had mentioned. If you are planning to add support, I would start by enabling ClickHouseDebeziumEmbeddedMongoIT .

[Ilanlido]

Why does it need the whole before/after structure?
if its an update / create it only needs the 'after' and the actual operator,
if its delete then it needs the 'before' to insert a new item with the same key (defined by order by?) with sign -1, am i wrong?

[aadant]

You can update the PK. In this case you need to delete / insert. Or imagine you change the partitioning key. We could optimize to detect that the PK / partitioning key did not change but it is an optimization and would be more error prone. Especially with schema overrides in CH, example different partitioning key.

[Ilanlido]

But the question is why would the sink currently expect BOTH 'before' and 'after' objects? i couldn't find a way in debezium to include both of them (without serializing) - which also controls the schema, or am i doing something wrong?

[subkanthi]

Actually for replacingmergetree u dont need both before and after, you just need the after, we had logic before to support collapsingmergetree where both were required.

[Ilanlido]

Got it thanks! therefore i might patch the 'expected schema' specifically for my case and expect to receive the actual object after the "io.debezium.transforms.ExtractChangedRecordState" smt

[suchak-dwivedi]

I am facing the very same problem. @Ilanlido I believe you meant to use "io.debezium.connector.mongodb.transforms.ExtractNewDocumentState". Did you get it to working state? If so, can you please share details.

[Ilanlido]

yeah the problem with using ExtractNewDocumentState is that it gets rid of the before / after envelope, since i'm not using CollapsingMergeTree and i don't care about the before object i manually patched the sink connector as if it received it with the envelope and continue as usual - hacky but works for me..

[suchak-dwivedi]

Can you share the snippet or point me to any fork you have?

[Ilanlido]

i don't have it publicly, the change is pretty small (and again hack as mentioned)
sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/converters/ClickHouseConverter.java

in convertStruct
i added this at the end of the function

        var operation = record.get(SinkRecordColumns.OPERATION).toString();

        if(operation.equalsIgnoreCase(CDC_OPERATION.DELETE.operation) || operation.equalsIgnoreCase(CDC_OPERATION.TRUNCATE.operation)) {
            record.put(SinkRecordColumns.BEFORE, struct);
        } else {
            record.put(SinkRecordColumns.AFTER, struct);
        }