Support topic-based encryption for Kafka buffer

[dlvenable]

Is your feature request related to a problem? Please describe.

For some situations, we want to encrypt each Kafka topic with a different encryption key.

Describe the solution you'd like

Update Data Prepper's Kafka buffer to support an optional encryption key.

buffer:
  kafka:
    topics:
      - name: MyTopic
         encryption_key: gEa68HffrNhFtJkNoY0UsD6D6W4w8KUFtYJTnte+eiY=

Additionally, this key could be encrypted by Amazon KMS so that we can support envelope encryption.

buffer:
  kafka:
    topics:
      - name: MyTopic
        encryption_key: AQIDAHhBQ4iH7RP28kWDRU1yN2K73qYEE2d8i06EBly7HoDSIwFXoO+oiW+HOlam8lfIUFwLAAAAfjB8BgkqhkiG9w0BBwagbzBtAgEAMGgGCSqGSIb3DQEHATAeBglghkgBZQMEAS4wEQQM/j9Uf9cxYv/poV0FAgEQgDuVG9jfls3Ys7dR/cRKmdkcYDJw/XzR/ZEnZwcT9e+XB1T+SxC0YHLtc33lRwoD/UV0Ot+y8oUBqMvaXg==
        kms_key_id: alias/ExampleAlias

Data Prepper initialization

When Data Prepper starts, it reads the pipeline configuration file. If the user provided a KMS key in the configuration, the Kafka Buffer decrypts the data encryption key using KMS.

Data Prepper holds the decrypted data key in memory for future processing.

Receiving data and writing to Kafka

As users provider their data to a Data Prepper source, that source writes to the Kafka buffer. The Kafka Buffer encrypts each record using the decrypted data key; this is the same key decrypted during initialization. The Kafka Buffer sends the encrypted record to the Kafka topic.

Processing data and reading from Kafka

As Data Prepper runs, the Pipeline Worker reads from the Kafka Buffer. The Kafka Buffer polls the Kafka topic for data. Each Kafka ConsumerRecord is already encrypted as described above. Thus, the Kafka Buffer decrypts this data using the same data encryption key it loaded at initialization.

Describe alternatives you've considered (Optional)**

N/A

Additional context

This builds on the work being done for #3322. The design could be extended for generic sink and source as well.

[dlvenable]

I updated the description to show the actual format. As with the serialization format, the encryption is now configured on the topic.