NPE when encrypting fields for SQLServer databases using debezium

[ktawfik]

I am using debezium to do a CDC from SQLServer to kafka, and as per the business needs, some of the columns must be encrypted.

Here is the snippet of the connector json file

{
"name": "live.sql.users",
...
        "transforms.unwrap.delete.handling.mode": "drop",
        "transforms": "unwrap,cipher",
        "predicates.isTombstone.type": "org.apache.kafka.connect.transforms.predicates.RecordIsTombstone",
        "transforms.unwrap.drop.tombstones": "false",
        "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
        "transforms.cipher.predicate": "isTombstone",
        "transforms.cipher.negate": "true",
        "transforms.cipher.cipher_data_keys": "[ { \"identifier\": \"my-key\", \"material\": { \"primaryKeyId\": 1000000001, \"key\": [ { \"keyData\": { \"typeUrl\": \"type.googleapis.com/google.crypto.tink.AesGcmKey\", \"value\": \"GhDLeulEJRDC8/19NMUXqw2jK\", \"keyMaterialType\": \"SYMMETRIC\" }, \"status\": \"ENABLED\", \"keyId\": 2000000002, \"outputPrefixType\": \"TINK\" } ] } } ]",
        "transforms.cipher.type": "com.github.hpgrahsl.kafka.connect.transforms.kryptonite.CipherField$Value",
        "transforms.cipher.cipher_mode": "ENCRYPT",
        "predicates": "isTombstone",
        "transforms.cipher.field_config": "[{\"name\":\"Password\"},{\"name\":\"MobNumber\"}, {\"name\":\"UserName\"}]",
        "transforms.cipher.cipher_data_key_identifier": "my-key"
...
}

and when I applied it, after few seconds I got the below error, when I call the /connectors/<connector_name>/status api

org.apache.kafka.connect.errors.ConnectException: Tolerance exceeded in error handler\n\t
at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:206)\n\t
at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execute(RetryWithToleranceOperator.java:132)\n\t
at org.apache.kafka.connect.runtime.TransformationChain.apply(TransformationChain.java:50)\n\t
at org.apache.kafka.connect.runtime.WorkerSourceTask.sendRecords(WorkerSourceTask.java:346)\n\t
at org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:261)\n\t
at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:191)\n\t
at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:240)\n\t
at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)\n\t
at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)\n\t
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)\n\t
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)\n\t
at java.base/java.lang.Thread.run(Thread.java:829)\nCaused by: org.apache.kafka.connect.errors.DataException: error: ENCRYPT of field path 'UserName' having data 'deleted605' failed unexpectedly\n\t
at com.github.hpgrahsl.kafka.connect.transforms.kryptonite.RecordHandler.processField(RecordHandler.java:90)\n\t
at com.github.hpgrahsl.kafka.connect.transforms.kryptonite.SchemaawareRecordHandler.lambda$matchFields$0(SchemaawareRecordHandler.java:73)\n\t
at java.base/java.util.ArrayList.forEach(ArrayList.java:1541)\n\t
at java.base/java.util.Collections$UnmodifiableCollection.forEach(Collections.java:1085)\n\t
at com.github.hpgrahsl.kafka.connect.transforms.kryptonite.SchemaawareRecordHandler.matchFields(SchemaawareRecordHandler.java:50)\n\t
at com.github.hpgrahsl.kafka.connect.transforms.kryptonite.CipherField.processWithSchema(CipherField.java:163)\n\t
at com.github.hpgrahsl.kafka.connect.transforms.kryptonite.CipherField.apply(CipherField.java:140)\n\t
at org.apache.kafka.connect.runtime.PredicatedTransformation.apply(PredicatedTransformation.java:56)\n\t
at org.apache.kafka.connect.runtime.TransformationChain.lambda$apply$0(TransformationChain.java:50)\n\t
at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndRetry(RetryWithToleranceOperator.java:156)\n\t
at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:190)\n\t
... 11 more\nCaused by: java.lang.NullPointerException\n\t
at com.esotericsoftware.kryo.util.DefaultGenerics.nextGenericTypes(DefaultGenerics.java:77)\n\t
at com.esotericsoftware.kryo.serializers.FieldSerializer.pushTypeVariables(FieldSerializer.java:144)\n\t
at com.esotericsoftware.kryo.serializers.FieldSerializer.write(FieldSerializer.java:102)\n\t
at com.esotericsoftware.kryo.Kryo.writeObject(Kryo.java:627)\n\t
at com.github.hpgrahsl.kafka.connect.transforms.kryptonite.RecordHandler.processField(RecordHandler.java:75)\n\t
... 21 more\n

Knowing that, the same configs working with other connectors with no problems

[ktawfik]

@hpgrahsl can you help in pointing what could be the issue?

[hpgrahsl]

hi @ktawfik thx for reaching out.

it's a bit difficult to point you to the actual issue based on the information you posted above. one thing that definitely cannot work in the configuration you provided is the key material you specified.

there is this line in the key config ... \"primaryKeyId\": 1000000001, ... and at the same time the keyset itself says ... \"keyId\": 2000000002, ... -> this cannot work but it should lead to a different exception not a NPE.

why you are getting the NPE here is not obvious to me for now. could you maybe run the connector with debug or even trace log level? then you get more useful log output. ideally you can also paste the full stacktrace you get.

[ktawfik]

Thanks @hpgrahsl for your quick reply, sorry for the typos, I was trying to change some of the actual values, but it fails :D, anyways, I was using a correct values and it used worked fine for days, but at some point in time, it gave the NPE, however I tried now to upgrade the version to use 0.4.0 instead of 0.3.1, and seems to be working fine, it is now working for just few hours, will see and let you know.

[hpgrahsl]

I see. Good to hear it's working again for now. In any case it would be interesting to learn more about your actual use case. Feel free to send me an email because it's always nice to hear about real-world usage :)

Also I'm closing this for now. If you run into problems related to this again feel free to re-open.

[ktawfik]

@hpgrahsl I appreciate if you can share with me your email to send you more details about my use case, as I can see some enhancements can be done to the library as well as I can see some documentation issues.

And the regarding this issue, it still exists, and it turned out to be a thread-safety issue with Kryo library (check this thread).

I can't find any pointers in the docs for kryptonite about the best practices to run multiple connectors using encryption, should I change the key specs or is it fine to use the same key (I think this is the issue with the thread-safety)?.

[hpgrahsl]

THX for sharing your findings so far. Regarding multi-threaded usage it's correct that there are currently no specific code components in there which would allow this out-of-the-box. It's something that can be added based on what the kryo documentation suggests, so the simple way is to just create multiple separate instances or e.g. with the pooling approach.

[hpgrahsl]

I forgot to mention with regard to your comment about improvements -> absolutely happy to receive any concrete input with details, even more appreciated if you open specific issues for the enhancements you have in mind no matter if it's related to the code, documentation or whatever - anything is welcome. happy to receive your pull requests for reviewing.

[hpgrahsl]

@ktawfik I gave it a quick closer look. and I'll work on a fix for this current issue during the next couple of days. there will be a patch release 0.4.1 and I'll keep you posted.

[ktawfik88]

@hpgrahsl perfect, very happy to know, I would like to have a look on the change, would be very beneficial, also for you to know, I opened a question on SoF and answered it for anyone later on face such problem, below is the link for your reference

https://stackoverflow.com/questions/75404421/npe-when-encrypting-fields-for-sqlserver-databases-using-debezium/75481901#75481901

[ktawfik88]

@hpgrahsl thanks for the quick turnaround, will test it and will let you know.

[hpgrahsl]

@ktawfik @ktawfik88 please try this patch release https://github.com/hpgrahsl/kryptonite-for-kafka/releases/tag/untagged-04c774210d245a928a64 and let me know if that helps to solve your issue.

[ktawfik88]

@hpgrahsl the patch url provides 404

[hpgrahsl]

was a draft release still. I published it now https://github.com/hpgrahsl/kryptonite-for-kafka/releases/tag/v0.4.1

[ktawfik88]

@hpgrahsl got it, deployed now and so far with the 10 connectors working fine, however it could take a day or 2 to see the race condition in action, will let you know. Thanks a lot 👍

[ktawfik88]

Hi @hpgrahsl, just wanted to thank you and confirm that the fix is working fine for the past 3 days, much appreciated 👍

[hpgrahsl]

nice. thx for reporting back. as mentioned previously I'd love to learn more about your actual use case of this project. you can DM me on twitter or email me at // grahslhp [ @ ] gmail.com //