JDBC Connector结合自定义加密的实现方式咨询

Great question! Let's break down your options based on exactly what you need to encrypt—whether it's entire messages or specific fields, and which approach fits best with your existing setup:

Option 1: Customize the Converter (Your Proposed OurCustomAvroConverter) #

If you need to encrypt the entire key or value payload before it hits Kafka (and decrypt the entire payload when reading from Kafka), then yes, replacing the default io.confluent.connect.avro.AvroConverter with a custom implementation makes perfect sense.

Here's how you'd approach it:

• Extend the existing AvroConverter class (so you don't have to reimplement all Avro serialization logic from scratch).
• Override the fromConnectData method (used by Source Connectors like JDBC to serialize data to Kafka) to encrypt the serialized Avro bytes before returning them.
• Override the toConnectData method (used by Sink Connectors to deserialize data from Kafka) to decrypt the bytes first, then pass them to the original Avro deserialization logic.

Your connector config would then look like this:

# Source Connector (JDBC) config
key.converter=com.yourcompany.OurCustomAvroConverter
key.converter.schema.registry.url=http://your-schema-registry:8081
value.converter=com.yourcompany.OurCustomAvroConverter
value.converter.schema.registry.url=http://your-schema-registry:8081
# Add your encryption configs (e.g., key store path, algorithm)
value.converter.encryption.key=your-encryption-key

This approach is clean because it ties encryption directly to the serialization/deserialization layer—any connector using this converter will automatically handle encryption/decryption, no extra config needed per connector.

Option 2: Use Custom Kafka Connect Transforms (Better for Field-Level Encryption) #

If you only need to encrypt specific sensitive fields (not the entire message), custom transforms are a more flexible choice. You won't need to modify the converter at all—instead, you'll add a transform step to your Source and Sink connector configs.

Source Connector (JDBC) Config Example (Encryption): #

transforms=encryptSensitiveField
transforms.encryptSensitiveField.type=com.yourcompany.EncryptField$Value
transforms.encryptSensitiveField.field.names=ssn,credit_card_number
transforms.encryptSensitiveField.encryption.algorithm=AES
transforms.encryptSensitiveField.encryption.key=your-key

Sink Connector Config Example (Decryption): #

transforms=decryptSensitiveField
transforms.decryptSensitiveField.type=com.yourcompany.DecryptField$Value
transforms.decryptSensitiveField.field.names=ssn,credit_card_number
transforms.decryptSensitiveField.encryption.algorithm=AES
transforms.decryptSensitiveField.encryption.key=your-key

Transforms operate on the Connect Data objects directly, so you can target exactly the fields you need without touching the underlying serialization format (Avro in your case). This is ideal if most of your message doesn't need encryption—you avoid unnecessary overhead and keep your converter logic clean.

Which Should You Choose? #

• Go with a custom converter if every part of your key/value needs encryption, or if you want encryption to be a universal part of how your data is serialized to Kafka.
• Go with custom transforms if you only need to protect specific fields, or if you want to keep encryption logic decoupled from the serialization layer.

Key Notes to Remember: #

• Never hardcode encryption keys in your configs—use a secure key management system or Kafka Connect's secure config providers to inject keys at runtime.
• Test performance: Encryption/decryption adds overhead, so benchmark your setup to ensure it doesn't bottleneck your data pipeline.
• Ensure your encryption/decryption logic is consistent between Source and Sink (same algorithm, key, padding, etc.) to avoid deserialization errors.

内容的提问来源于Stack Exchange，提问作者SL101