KAFKA-470: Add stateless field decryption and add MongoDB FLE support by mdb-gtenrreiro · Pull Request #204 · mongodb/mongo-kafka

mdb-gtenrreiro · 2026-03-10T13:42:12Z

No description provided.

Implement two encryption features for Oracle-to-MongoDB migration: 1. Pluggable Field Decryption - New FieldValueTransformer interface for custom decryption logic - FieldValueTransformPostProcessor recursively transforms configured fields - SampleAesFieldValueTransformer as reference implementation - Configuration: field.value.transformer, field.value.transformer.fields, etc. - Auto-registers in PostProcessor chain when configured 2. Client-Side Field Level Encryption (CS-FLE) - Enable MongoDB native CS-FLE in Sink Connector - MongoSinkTask builds AutoEncryptionSettings from configuration - Configuration: csfle.enabled, csfle.key.vault.namespace, csfle.local.master.key, csfle.schema.map - Added mongodb-crypt:5.6.4 dependency Test Coverage: - 28 new tests (all passing, 645 total) - 15 tests for FieldValueTransformPostProcessor - 8 tests for SampleAesFieldValueTransformer - 8 tests for CS-FLE configuration - 3 integration tests for PostProcessor chain Demo: - Comprehensive demo in demos/field-decryption-and-csfle/ - Automated run-demo.sh script - Shows Oracle AES decryption + MongoDB CS-FLE re-encryption - Uses Confluent Platform images for x86_64 compatibility - Includes detailed README and implementation summary Files Modified: - build.gradle.kts - MongoSinkConfig.java - MongoSinkTopicConfig.java - MongoSinkTask.java - PostProcessors.java - MongoSinkConfigTest.java Files Added: - FieldValueTransformer.java (interface) - FieldValueTransformPostProcessor.java - SampleAesFieldValueTransformer.java - FieldValueTransformPostProcessorTest.java - SampleAesFieldValueTransformerTest.java - MongoSinkTaskCsfleTest.java - demos/field-decryption-and-csfle/ (complete demo)

arahmanan · 2026-03-12T12:40:46Z

src/main/java/com/mongodb/kafka/connect/sink/MongoSinkConfig.java

+        CSFLE_KEY_VAULT_NAMESPACE_DISPLAY);
+    configDef.define(
+        CSFLE_LOCAL_MASTER_KEY_CONFIG,
+        Type.STRING,


Let's use Type.Password here instead

From MSK's docs:

To prevent secrets from appearing in connector log files, a plugin developer must use the Kafka Connect enum constant ConfigDef.Type.PASSWORD to define sensitive properties. When a property is type ConfigDef.Type.PASSWORD, Kafka Connect excludes its value from connector logs even if the value is sent as plaintext.

arahmanan · 2026-03-12T12:51:30Z

Design Proposal - MongoDB Kafka Sink Connector - Field Level Encrypt_Decrypt.txt

@@ -0,0 +1,484 @@
+Design Proposal - MongoDB Kafka Sink Connector 


do we still need this file?

arahmanan · 2026-03-12T12:54:49Z

build.gradle.kts

+    options.isDebug = true
+    options.debugOptions.debugLevel = "source,lines,vars"


let's remove these

arahmanan · 2026-03-12T13:15:50Z

build.gradle.kts

    implementation("org.apache.avro:avro:${project.extra["avroVersion"]}")

    mongoAndAvroDependencies("org.mongodb:mongodb-driver-sync:${project.extra["mongodbDriverVersion"]}")
+    mongoAndAvroDependencies("org.mongodb:mongodb-crypt:1.11.0")


what's the reason for using an older version? Can we use the latest here?

11.1.0 seems to be the latest version that is compatible with mongodb driver 4.7

arahmanan

still going through this 😅 Sharing some more comments for now

arahmanan · 2026-03-13T17:57:42Z

demos/field-decryption-and-csfle/IMPLEMENTATION-SUMMARY.md

@@ -0,0 +1,123 @@
+# Implementation Summary


seems like an outdated duplicate of the README. Should we remove it?

I removed that file and added an implementation section to the README.md. Not sure we need that but it seemed helpful.

arahmanan · 2026-03-13T18:01:50Z

demos/field-decryption-and-csfle/demo-encryption-with-csfle.sh

+info "Cleanup commands:"
+echo "  curl -X DELETE $CONNECT_URL/connectors/enc-source"
+echo "  curl -X DELETE $CONNECT_URL/connectors/csfle-sink"
+echo "  cd demo && docker compose down -v"


[opt] newline at the end of this file

added a new line

arahmanan · 2026-03-13T20:48:02Z

src/main/java/com/mongodb/kafka/connect/sink/MongoSinkTask.java

  }
+
+  @VisibleForTesting(otherwise = PRIVATE)
+  static AutoEncryptionSettings buildAutoEncryptionSettings(final MongoSinkConfig sinkConfig) {


can we add an integration test that confirms we're encrypting the correct fields? Maybe the demos can be turned into integration tests

Added tests for this; I removed the demos.

I personally think demo examples can be helpful for complex functionality like this, but I replaced them with tests to address the review feedback.

arahmanan · 2026-03-13T21:07:08Z

...m/mongodb/kafka/connect/sink/processor/field/transform/FieldValueTransformPostProcessor.java

+    super(config);
+    this.transformer = createTransformer(config);
+    String fieldsCsv = config.getString(FIELD_VALUE_TRANSFORMER_FIELDS_CONFIG);
+    this.targetFields = new HashSet<>(Arrays.asList(fieldsCsv.split("\\s*,\\s*")));


can we add a unit test to make sure this works properly with more than 2 fields and subfields? It might actually be best to have an integration test that confirms we properly support encrypted subfields

Added tests

arahmanan · 2026-03-13T22:09:06Z

...m/mongodb/kafka/connect/sink/processor/field/transform/FieldValueTransformPostProcessor.java

+    this.transformer = createTransformer(config);
+    String fieldsCsv = config.getString(FIELD_VALUE_TRANSFORMER_FIELDS_CONFIG);
+    this.targetFields = new HashSet<>(Arrays.asList(fieldsCsv.split("\\s*,\\s*")));
+    this.failOnError = config.getBoolean(FIELD_VALUE_TRANSFORMER_FAIL_ON_ERROR_CONFIG);


we already have mongo.errors.tolerance (docs). Should we use that instead?
We should also test this out with a dead letter queue setup and make sure we aren't adding unencrypted data to the DLQ. E.g. what happens if we successfully decrypt the first field and fail to decrypt the second one? Would we end up sending to the DLQ and unencrypted field?

You are right about the "tolerance" config. I removed the redundant configuration change.

The messages going the DLQ were already free from unencrypted data. I added a new integration test to make sure not data is leaked.

arahmanan · 2026-03-13T22:43:47Z

src/main/java/com/mongodb/kafka/connect/sink/MongoSinkTask.java

+            .keyVaultNamespace(keyVaultNamespace)
+            .kmsProviders(kmsProviders)
+            .extraOptions(extraOptions)
+            .bypassQueryAnalysis(true);


Disables automatic analysis of outgoing commands. Specify true to use explicit encryption without the Automatic Encryption Shared Library.

Does this mean we need to explicitly encrypt? I guess the integration tests I suggested adding will confirm if this is an issue.

It had to be explicit before; but now I ve changed it so that it is configurable so that when:

bypassQueryAnalysis=false

automatic encryptions works correctly

and when:

bypassQueryAnalysis=true

automatic encryption is DISABLED and it would need explicit encryption.

- Consolidated implementation details into the main demo README - Added architecture details for Field Decryption and CS-FLE - Included configuration properties reference - Added known limitations section - Removed redundant IMPLEMENTATION-SUMMARY.md file

semgrep-code-mongodb · 2026-03-16T02:27:26Z

src/integrationTest/java/com/mongodb/kafka/connect/sink/FieldDecryptionIntegrationTest.java

+  private String encryptAes(final String plaintext) {
+    try {
+      SecretKeySpec secretKey = new SecretKeySpec(AES_KEY.getBytes(StandardCharsets.UTF_8), "AES");
+      Cipher cipher = Cipher.getInstance("AES/ECB/PKCS5Padding");


Use of AES with ECB mode detected. ECB doesn't provide message confidentiality and is not semantically secure so should not be used. Instead, use a strong, secure cipher: Cipher.getInstance("AES/CBC/PKCS7PADDING"). See https://owasp.org/www-community/Using_the_Java_Cryptographic_Extensions for more information.

🎉 Removed in commit 196e587 🎉

/other This is test-only code used to simulate legacy AES encryption for integration testing and demos. The ECB mode is intentionally used to demonstrate decrypting data from legacy systems that may use insecure encryption. This code is never used in production - it only exists in integration tests and demo scripts to create sample encrypted data that the connector then decrypts.

semgrep-code-mongodb · 2026-03-16T02:27:27Z

src/integrationTest/java/com/mongodb/kafka/connect/sink/FieldDecryptionIntegrationTest.java

+  private String encryptAes(final String plaintext) {
+    try {
+      SecretKeySpec secretKey = new SecretKeySpec(AES_KEY.getBytes(StandardCharsets.UTF_8), "AES");
+      Cipher cipher = Cipher.getInstance("AES/ECB/PKCS5Padding");


Cipher in ECB mode is detected. ECB mode produces the same output for the same input each time which allows an attacker to intercept and replay the data. Further, ECB mode does not provide any integrity checking. See https://find-sec-bugs.github.io/bugs.htm#CIPHER_INTEGRITY.

🚀 Removed in commit 196e587 🚀

/other This is test-only code used to simulate legacy AES encryption for integration testing and demos. The ECB mode is intentionally used to demonstrate decrypting data from legacy systems that may use insecure encryption. This code is never used in production - it only exists in integration tests and demo scripts to create sample encrypted data that the connector then decrypts.

Calvinnix · 2026-03-16T14:38:47Z

demos/field-decryption/run-demo.sh

+step "Demo Complete!"
+
+info "To stop the demo:"
+echo "  cd $SCRIPT_DIR && docker compose down -v"


I like the demo experience but I'm not sure this should live in this repo, perhaps we could look into moving this into the kafka-edu repo or maybe a net new repo dedicated to demos.

yeah you might be right. Let me ping Product and see what they want.

Calvinnix · 2026-03-16T14:55:16Z

src/integrationTest/java/com/mongodb/kafka/connect/mongodb/MongoKafkaTestCase.java

              }
            });
+    // Clean up any stale JMX MBeans from previous test runs
+    cleanUpMBeans();


I'm guessing this is a driveby that addresses flaky tests?

Yes I had issues with flakyness and this seems to quiet it down

src/integrationTest/java/com/mongodb/kafka/connect/sink/CsfleIntegrationTest.java

Calvinnix · 2026-03-16T16:23:34Z

src/main/java/com/mongodb/kafka/connect/sink/MongoSinkConfig.java

+  public static final String CSFLE_BYPASS_QUERY_ANALYSIS_CONFIG = "csfle.bypass.query.analysis";
+  private static final String CSFLE_BYPASS_QUERY_ANALYSIS_DOC =
+      "Whether to bypass automatic query analysis for CS-FLE. "
+          + "When true, automatic encryption based on schema maps is disabled and explicit encryption must be used. "
+          + "When false, automatic encryption is enabled but requires mongocryptd or crypt_shared library. "
+          + "Default is true to avoid external dependencies.";
+  private static final boolean CSFLE_BYPASS_QUERY_ANALYSIS_DEFAULT = true;
+  private static final String CSFLE_BYPASS_QUERY_ANALYSIS_DISPLAY = "CS-FLE bypass query analysis";


I'm a bit confused about this logic, I didn't see a test that tested this being true either, could you add a test? That may shed some light on this.

Added tests here:

mongo-kafka/src/integrationTest/java/com/mongodb/kafka/connect/sink/CsfleIntegrationTest.java

Line 293 in 76ad4c5

@DisplayName("Ensure bypass query analysis disables automatic encryption")

and Here:

mongo-kafka/src/test/java/com/mongodb/kafka/connect/sink/MongoSinkTaskCsfleTest.java

Line 172 in 76ad4c5

@DisplayName("test bypass query analysis defaults to true")

Calvinnix · 2026-03-16T17:38:53Z

...com/mongodb/kafka/connect/sink/processor/field/transform/SampleAesFieldValueTransformer.java

+
+  public static final String AES_KEY_CONFIG = "field.value.transformer.aes.key";
+  public static final String AES_ALGORITHM_CONFIG = "field.value.transformer.aes.algorithm";
+  private static final String DEFAULT_ALGORITHM = "AES/ECB/PKCS5Padding";


The bot detected this in the tests but I think it is probably worth defaulting to a more secure algorithm due to the concerns around ECB that the bot highlighted.

Even though this is a sample it appears to be fully functional so it could still be used technically.

Calvinnix · 2026-03-16T19:03:07Z

src/main/java/com/mongodb/kafka/connect/sink/MongoSinkConfig.java

+      "The key vault namespace for CS-FLE in the format 'database.collection'. "
+          + "Example: 'encryption.__keyVault'.";
+  private static final String CSFLE_KEY_VAULT_NAMESPACE_DEFAULT = "";
+  private static final String CSFLE_KEY_VAULT_NAMESPACE_DISPLAY = "CS-FLE Key vault namespace";


We should add validation to make sure these are setup as expected when csfle.enabled is true.
something like:

if ("true".equalsIgnoreCase(props.getOrDefault(CSFLE_ENABLED_CONFIG, "false"))) { if (props.getOrDefault(CSFLE_KEY_VAULT_NAMESPACE_CONFIG, "").isEmpty()) { results.get(CSFLE_KEY_VAULT_NAMESPACE_CONFIG) .addErrorMessage("csfle.key.vault.namespace must be set when csfle.enabled=true"); } if (props.getOrDefault(CSFLE_LOCAL_MASTER_KEY_CONFIG, "").isEmpty()) { results.get(CSFLE_LOCAL_MASTER_KEY_CONFIG) .addErrorMessage("csfle.local.master.key must be set when csfle.enabled=true"); } }

Added extra validation

Calvinnix · 2026-03-16T19:04:35Z

src/main/java/com/mongodb/kafka/connect/sink/MongoSinkTask.java

+            .extraOptions(extraOptions);
+
+    if (bypassQueryAnalysis) {
+      autoEncryptionBuilder.bypassQueryAnalysis(true);


why do we need to set this twice?

e.g. set already here

if (bypassQueryAnalysis) { extraOptions.put("bypassQueryAnalysis", true); }

nice catch, remove the redundant one

Calvinnix · 2026-03-16T19:08:01Z

...com/mongodb/kafka/connect/sink/processor/field/transform/SampleAesFieldValueTransformer.java

+      byte[] encryptedBytes = Base64.getDecoder().decode(encrypted);
+      Cipher cipher = Cipher.getInstance(algorithm);
+      cipher.init(Cipher.DECRYPT_MODE, secretKey);


This runs for each document, shouldn't this be in the init?

mdb-gtenrreiro added 3 commits March 9, 2026 16:17

KAFKA-470

1e499ab

KAFKA-470

510b014

mdb-gtenrreiro requested a review from a team as a code owner March 10, 2026 13:42

mdb-gtenrreiro requested a review from mdb-vpurohit March 10, 2026 13:42

mdb-gtenrreiro marked this pull request as draft March 10, 2026 13:42

mdb-gtenrreiro added 9 commits March 10, 2026 14:28

KAFKA-470

b2ba946

KAFKA-470

18e0100

KAFKA-470

ec52e76

KAFKA-470

e5226dc

KAFKA-470

ca37912

KAFKA-470

f09c23a

KAFKA-470

05d9d3b

KAFKA-470

3ec7d8f

KAFKA-470

0d9bdc1

mdb-gtenrreiro marked this pull request as ready for review March 12, 2026 00:43

KAFKA-470

d86de85

mdb-gtenrreiro requested review from Calvinnix and arahmanan March 12, 2026 01:26

arahmanan reviewed Mar 12, 2026

View reviewed changes

mdb-gtenrreiro added 3 commits March 12, 2026 18:36

KAFKA-470

1326f3d

KAFKA-470

87178cf

KAFKA-470

875dd67

mdb-gtenrreiro requested a review from arahmanan March 12, 2026 18:47

Attempt ot fix flaky test

e0d104e

arahmanan reviewed Mar 13, 2026

View reviewed changes

mdb-gtenrreiro added 3 commits March 16, 2026 00:56

KAFKA-470

7a2d9db

KAFKA-470

3b9acff

semgrep-code-mongodb bot reviewed Mar 16, 2026

View reviewed changes

mdb-gtenrreiro requested a review from arahmanan March 16, 2026 02:33

KAFKA-470

bc3d1f0

Calvinnix reviewed Mar 16, 2026

View reviewed changes

KAFKA-470

76ad4c5

Calvinnix requested changes Mar 16, 2026

View reviewed changes

mdb-gtenrreiro added 3 commits March 16, 2026 20:41

KAFKA-470

11c6e05

KAFKA-470

196e587

KAFKA-470

a903a96

Calvinnix removed the request for review from mdb-vpurohit March 16, 2026 20:54

mdb-gtenrreiro added 5 commits March 16, 2026 21:02

KAFKA-470

dd7c892

KAFKA-470

6478f28

KAFKA-470

be7a0b0

KAFKA-470

b774b2c

KAFKA-470

e454ba9

mdb-gtenrreiro requested a review from Calvinnix March 17, 2026 02:03

		@@ -0,0 +1,484 @@
		Design Proposal - MongoDB Kafka Sink Connector

		options.isDebug = true
		options.debugOptions.debugLevel = "source,lines,vars"

Conversation

mdb-gtenrreiro commented Mar 10, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

arahmanan left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mdb-gtenrreiro Mar 16, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

semgrep-code-mongodb bot Mar 16, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

semgrep-code-mongodb bot Mar 16, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

mdb-gtenrreiro Mar 16, 2026 •

edited

Loading

semgrep-code-mongodb bot Mar 16, 2026 •

edited

Loading

semgrep-code-mongodb bot Mar 16, 2026 •

edited

Loading