Repository: kroxylicious/kroxylicious
Pull request #696 (2023-10-25 02:14:43)

Initial work on an Envelope Encryption filter
This is decomposed into a number of modules.

* kms defines the abstract API for a KMS that supports envelope encryption and an in-memory implementation for test purposes
* Other kms-provider-* modules will (eventually) implement the KMS API and be discoverable as services
* The kroxylicious-encryption module uses the API for KMS to do envelope encryption of records headed into a cluster.

Obviously there are a million TODOs here, and the decryption part is not yet implemented (!), but hopefully then general shape is apparent.

### Type of change

_Select the type of your PR_

- Bugfix
- Enhancement / new feature
- Refactoring
- Documentation

### Description

_Please describe your pull request_

### Additional Context

_Why are you making this pull request?_

### Checklist

_Please go through this checklist and make sure all applicable tasks have been done_

- [ ] Write tests
- [ ] Make sure all tests pass
- [ ] Review performance test results. Ensure that any degradations to performance numbers are understood and justified.
- [ ] Make sure all Sonar-Lint warnings are addressed or are justifiably ignored.
- [ ] Update documentation
- [ ] Reference relevant issue(s) and close them after merging
- [ ] For user facing changes, update CHANGELOG.md (remember to include changes affecting the API of the test artefacts too).


k-wall: I thought the intent was to go HashiCorp vault only to begin with, so I was mildly surprised to see AWS etc.  
tombentley: > I thought the intent was to go HashiCorp vault only to begin with, so I was mildly surprised to see AWS etc. This is not an objection.

Yeah, this is just scaffolding. I can delete those modules if you like.
tombentley: > I was surprised to see consumers/producers. I thought the two kroxy replicas A and B using KEK K to encrypt topic T would each call the KMS to create their own DEKs. The fact that they'd be using different DEKs is of no consequence. At least in my mind.

That way does work. But it has some drawbacks. Reading some of the NIST recommendations last week made me think about alternatives. Let me start a design discussion about this.
k-wall: > > I thought the intent was to go HashiCorp vault only to begin with, so I was mildly surprised to see AWS etc. This is not an objection.
> 
> Yeah, this is just scaffolding. I can delete those modules if you like.

I don't feel strongly.  Implementing earlier will help us know if the API abstraction is sufficiently general. 
k-wall: > > I was surprised to see consumers/producers. I thought the two kroxy replicas A and B using KEK K to encrypt topic T would each call the KMS to create their own DEKs. The fact that they'd be using different DEKs is of no consequence. At least in my mind.
> 
> That way does work. But it has some drawbacks. Reading some of the NIST recommendations last week made me think about alternatives. Let me start a design discussion about this.

I suppose sharing the DEK across replicas means reduced the number of times the KEK is asked to produce a DEK.  That has the advantage the the need to rotate so frequently the KEK is diminished.   That's all true, but I think there's the option to defer implementation (until after after a initial release).

tombentley: > > > I was surprised to see consumers/producers. I thought the two kroxy replicas A and B using KEK K to encrypt topic T would each call the KMS to create their own DEKs. The fact that they'd be using different DEKs is of no consequence. At least in my mind.
> > 
> > 
> > That way does work. But it has some drawbacks. Reading some of the NIST recommendations last week made me think about alternatives. Let me start a design discussion about this.

Replying to myself...

I studied some of the NIST recommendations more closely last week. Specifically

* [SP.800-130: A Framework for Designing Cryptographic Key Management Systems](https://csrc.nist.gov/pubs/sp/800/130/final)
* [SP.800-152: A Profile for U.S. Federal Cryptographic Key Management Systems (CKMS)](https://csrc.nist.gov/pubs/sp/800/152/final)

SP.800-130 says:

> A CKMS includes all devices or sub-systems that can access an unencrypted key or its metadata. 

Because a filter implementing envelope encryption necessarily accesses the decrypted DEK it follows that _we are in the business of writing a CKMS_. 

But on re-reading, a lot the requirements in those documents are on the _design documentation_, and don't necessarily imply requirements on the system itself.  So, for example, the system doesn't have to maintain metadata on all the DEKs that it uses, but the design documentation does need to describe what metadata we do keep. The situation is complicated by the fact that the overall system involves a separate KMS for the KEK and the envelope encryption filter, which is a KMS for DEKs.  This situation seems to imply that we'd need separate "design documentation" for each KMS we integrate with if someone wants to assess it against the criteria in the above documents. Therefore we're possibly better off following an implementation strategy where we thoroughly integrate with each KMS in turn, and generate the documentation required, rather than an approach where we just "make it work" with each KMSs API. 

Anyway, I'm digressing a bit... 

I'm now thinking that, if we don't really have to have metadata about DEKs then we should aim for the "all-in-band" book keeping mechanism, at least initially. I _think_ the record inflation problem mostly goes away if we use batch compression, because all the encrypted records in a batch will have the same prefix (kek id, dek), so should compress well.

I also took a look at [NIST SP.800-38D](https://csrc.nist.gov/pubs/sp/800/38/d/final) which goes into the gory details of IV generation for AES-GCM. 

* My reading of it is that we should generate 96-bit IVs using the non-deterministic method, incrementing the "random field". This will mean the IVs compress well too because the leading bytes will be the same for each record in the batch. 
* Note that this bring requirements on how many times we can safely use a DEK with this IV generation scheme.
* But that's fine under the all-in-band book keeping method because it's relatively easy for a proxy instance to track how many times a DEK has been used and to request a new one if needed. 

-------

Now replying to @k-wall:
 
> I suppose sharing the DEK across replicas means reduced the number of times the KEK is asked to produce a DEK. That has the advantage the the need to rotate so frequently the KEK is diminished. That's all true, but I think there's the option to defer implementation (until after after a initial release).

The key rotation for KEK is _mostly_ in the KMS's domain. In particular they all seem to support at least time-based key rotation. And we've made the decision to assume pre-provisioned KEKs for now. One thing that we should think more about is whether/how we deal with generating new DEKs when a KEK is rotated in the KMS. In other words we need to handle:

1. DEK rotation based on number of records it has encrypted (at least if using AES-GCM and the non-deterministic IV generation proposed above).
2. DEK rotation based on the expiry of the KEK. 
tombentley: @k-wall I've now written tests for the in-memory KMS and documented the API in the `kms` module better, so PTAL at this specifically. 
sonarcloud[bot]: SonarCloud Quality Gate failed.&nbsp; &nbsp; [![Quality Gate failed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/failed-16px.png 'Quality Gate failed')](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=696)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG) [![C](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/C-16px.png 'C')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG) [32 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG)  
[![Vulnerability](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/vulnerability-16px.png 'Vulnerability')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY)  
[![Security Hotspot](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/security_hotspot-16px.png 'Security Hotspot')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT)  
[![Code Smell](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/code_smell-16px.png 'Code Smell')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL) [37 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL)

[![27.2%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/25-16px.png '27.2%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_coverage&view=list) [27.2% Coverage](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_coverage&view=list)  
[![0.0%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/Duplications/3-16px.png '0.0%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_duplicated_lines_density&view=list)



![idea](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/light_bulb-16px.png 'idea') Catch issues before they fail your Quality Gate with our IDE extension ![sonarlint](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/sonarlint-16px.png 'sonarlint') [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=sonarcloud-welcome)
sonarcloud[bot]: SonarCloud Quality Gate failed.&nbsp; &nbsp; [![Quality Gate failed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/failed-16px.png 'Quality Gate failed')](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=696)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG) [![C](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/C-16px.png 'C')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG) [3 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG)  
[![Vulnerability](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/vulnerability-16px.png 'Vulnerability')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY)  
[![Security Hotspot](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/security_hotspot-16px.png 'Security Hotspot')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT)  
[![Code Smell](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/code_smell-16px.png 'Code Smell')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL) [17 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL)

[![30.5%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/25-16px.png '30.5%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_coverage&view=list) [30.5% Coverage](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_coverage&view=list)  
[![0.0%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/Duplications/3-16px.png '0.0%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_duplicated_lines_density&view=list)



![idea](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/light_bulb-16px.png 'idea') Catch issues before they fail your Quality Gate with our IDE extension ![sonarlint](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/sonarlint-16px.png 'sonarlint') [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=sonarcloud-welcome)
sonarcloud[bot]: SonarCloud Quality Gate failed.&nbsp; &nbsp; [![Quality Gate failed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/failed-16px.png 'Quality Gate failed')](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=696)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG) [![C](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/C-16px.png 'C')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG) [3 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG)  
[![Vulnerability](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/vulnerability-16px.png 'Vulnerability')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY)  
[![Security Hotspot](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/security_hotspot-16px.png 'Security Hotspot')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT)  
[![Code Smell](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/code_smell-16px.png 'Code Smell')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL) [17 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL)

[![47.6%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/40-16px.png '47.6%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_coverage&view=list) [47.6% Coverage](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_coverage&view=list)  
[![0.0%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/Duplications/3-16px.png '0.0%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_duplicated_lines_density&view=list)



![idea](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/light_bulb-16px.png 'idea') Catch issues before they fail your Quality Gate with our IDE extension ![sonarlint](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/sonarlint-16px.png 'sonarlint') [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=sonarcloud-welcome)
sonarcloud[bot]: SonarCloud Quality Gate failed.&nbsp; &nbsp; [![Quality Gate failed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/failed-16px.png 'Quality Gate failed')](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=696)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG) [![C](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/C-16px.png 'C')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG) [4 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG)  
[![Vulnerability](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/vulnerability-16px.png 'Vulnerability')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY)  
[![Security Hotspot](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/security_hotspot-16px.png 'Security Hotspot')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT)  
[![Code Smell](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/code_smell-16px.png 'Code Smell')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL) [18 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL)

[![52.9%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/50-16px.png '52.9%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_coverage&view=list) [52.9% Coverage](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_coverage&view=list)  
[![0.0%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/Duplications/3-16px.png '0.0%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_duplicated_lines_density&view=list)



![idea](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/light_bulb-16px.png 'idea') Catch issues before they fail your Quality Gate with our IDE extension ![sonarlint](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/sonarlint-16px.png 'sonarlint') [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=sonarcloud-welcome)
k-wall: @tombentley great work on both APIs and implementation.
tombentley: @k-wall thanks for the detailed review, I really appreciate it, some great points there! I only managed to look at some of the comments today, I'll try to pick up the rest tomorrow.

tombentley: > Shouldn't the DekContext know when its being used to encrypt something and thus track its own usage? It could then refuse to encrypt /throw past expiry and get users to regenerate?

[not sure why github wouldn't let me reply to this one in-line...]

Something should care, yes... I have a refactoring locally which I'll push once I've processed more of Keith's feedback. That refactor makes the `DekCache` more of a `KeyManager` (and `DekContext` → `KeyContext`). Probably the context keep a count, but the manager handles expiry. I.e. the context is dumb, and the manager applies the rules.
sonarcloud[bot]: SonarCloud Quality Gate failed.&nbsp; &nbsp; [![Quality Gate failed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/failed-16px.png 'Quality Gate failed')](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=696)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG) [![C](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/C-16px.png 'C')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG) [3 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG)  
[![Vulnerability](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/vulnerability-16px.png 'Vulnerability')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY)  
[![Security Hotspot](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/security_hotspot-16px.png 'Security Hotspot')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT)  
[![Code Smell](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/code_smell-16px.png 'Code Smell')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL) [26 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL)

[![60.8%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/60-16px.png '60.8%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_coverage&view=list) [60.8% Coverage](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_coverage&view=list)  
[![0.0%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/Duplications/3-16px.png '0.0%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_duplicated_lines_density&view=list)



![idea](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/light_bulb-16px.png 'idea') Catch issues before they fail your Quality Gate with our IDE extension ![sonarlint](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/sonarlint-16px.png 'sonarlint') [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=sonarcloud-welcome)
k-wall: @tombentley I'm adding more tests here: https://github.com/tombentley/kproxy/pull/4
sonarcloud[bot]: SonarCloud Quality Gate failed.&nbsp; &nbsp; [![Quality Gate failed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/failed-16px.png 'Quality Gate failed')](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=696)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG) [![C](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/C-16px.png 'C')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG) [3 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG)  
[![Vulnerability](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/vulnerability-16px.png 'Vulnerability')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY)  
[![Security Hotspot](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/security_hotspot-16px.png 'Security Hotspot')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT)  
[![Code Smell](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/code_smell-16px.png 'Code Smell')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL) [27 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL)

[![60.6%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/60-16px.png '60.6%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_coverage&view=list) [60.6% Coverage](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_coverage&view=list)  
[![0.0%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/Duplications/3-16px.png '0.0%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_duplicated_lines_density&view=list)



![idea](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/light_bulb-16px.png 'idea') Catch issues before they fail your Quality Gate with our IDE extension ![sonarlint](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/sonarlint-16px.png 'sonarlint') [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=sonarcloud-welcome)
k-wall: @tombentley I pushed to https://github.com/tombentley/kproxy/pull/4.  Its mainly ITs with a defect fix.
sonarcloud[bot]: SonarCloud Quality Gate failed.&nbsp; &nbsp; [![Quality Gate failed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/failed-16px.png 'Quality Gate failed')](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=696)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG) [![C](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/C-16px.png 'C')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG) [2 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG)  
[![Vulnerability](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/vulnerability-16px.png 'Vulnerability')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY)  
[![Security Hotspot](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/security_hotspot-16px.png 'Security Hotspot')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT)  
[![Code Smell](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/code_smell-16px.png 'Code Smell')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL) [27 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL)

[![60.8%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/60-16px.png '60.8%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_coverage&view=list) [60.8% Coverage](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_coverage&view=list)  
[![0.0%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/Duplications/3-16px.png '0.0%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_duplicated_lines_density&view=list)



![idea](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/light_bulb-16px.png 'idea') Catch issues before they fail your Quality Gate with our IDE extension ![sonarlint](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/sonarlint-16px.png 'sonarlint') [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=sonarcloud-welcome)
k-wall: One more PR raised against your branch: https://github.com/tombentley/kproxy/pull/5
tombentley: @k-wall this now includes your recent changes. 
sonarcloud[bot]: SonarCloud Quality Gate failed.&nbsp; &nbsp; [![Quality Gate failed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/failed-16px.png 'Quality Gate failed')](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=696)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG) [![C](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/C-16px.png 'C')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG) [2 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG)  
[![Vulnerability](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/vulnerability-16px.png 'Vulnerability')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY)  
[![Security Hotspot](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/security_hotspot-16px.png 'Security Hotspot')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT)  
[![Code Smell](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/code_smell-16px.png 'Code Smell')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL) [25 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL)

[![61.2%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/60-16px.png '61.2%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_coverage&view=list) [61.2% Coverage](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_coverage&view=list)  
[![0.0%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/Duplications/3-16px.png '0.0%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_duplicated_lines_density&view=list)



![idea](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/light_bulb-16px.png 'idea') Catch issues before they fail your Quality Gate with our IDE extension ![sonarlint](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/sonarlint-16px.png 'sonarlint') [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=sonarcloud-welcome)
sonarcloud[bot]: SonarCloud Quality Gate failed.&nbsp; &nbsp; [![Quality Gate failed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/failed-16px.png 'Quality Gate failed')](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=696)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG)  
[![Vulnerability](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/vulnerability-16px.png 'Vulnerability')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY)  
[![Security Hotspot](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/security_hotspot-16px.png 'Security Hotspot')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT)  
[![Code Smell](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/code_smell-16px.png 'Code Smell')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL) [30 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL)

[![59.1%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/50-16px.png '59.1%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_coverage&view=list) [59.1% Coverage](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_coverage&view=list)  
[![0.0%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/Duplications/3-16px.png '0.0%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_duplicated_lines_density&view=list)



![idea](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/light_bulb-16px.png 'idea') Catch issues before they fail your Quality Gate with our IDE extension ![sonarlint](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/sonarlint-16px.png 'sonarlint') [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=sonarcloud-welcome)
tombentley: @k-wall I've added some functionality for actually caching keys in the key manager. This helps to avoid the pattern where every request involves a call to the KMS (either to generate a new DEK or to decrypt an eDEK ). Right now it has resulted in a lot of TODOs because it's not obvious what the right strategies are for things like expiring DEKs from the cache. There is an inherent trade-off between minimising calls to the KMS (which will help latency on fetch and produce paths) and having a lot of key material in memory. With this change we've basically moved from one extreme to the other. I think we need some way to say "keep up to N DEKs in the cache and remove DEKs which haven't been used for T seconds".

Separately from this I've been thinking about what's the best way to deal with rotating DEKs. With these changes we keep a count of the number of times a DEK has been used. Right now I'm kind-of using a pattern where we look at the number of records to be encrypted and use the cached key if that wouldn't exceed the allowed number. That doesn't really work well for a few reasons:

1. It might get gnarly when we solve the threading problem. I'm using atomic CAS for this, but I've not really convinced myself that it's correct.
2. We don't always know (efficiently) how many records there are in a batch, so this kind of "accounting ahead of time" is inefficient for older message formats (maybe not a common problem in practice)
3. It doesn't (won't) account correctly for null values which aren't (well, won't) be encryted.
4. It makes for an infinite loop if the number of records in the stream is larger than the number of records allowed by policy.  

The obviously alternative would be to request a new DEK only once we've definitely exhausted the current one, which would solve 2, 3 and 4 but add latency (unless we tried to get clever with pre-generating DEKs, but that sees like way too much complexity right now). So I'm interested in your thoughts about this.


tombentley: Oh, also @k-wall WDYT about splitting out the KMS part of this (assuming you're happy with it) into a separate PR for merging in the short term, while we continue to hack around with the actual encryption filter part in draft PRs?
sonarcloud[bot]: SonarCloud Quality Gate failed.&nbsp; &nbsp; [![Quality Gate failed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/failed-16px.png 'Quality Gate failed')](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=696)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG)  
[![Vulnerability](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/vulnerability-16px.png 'Vulnerability')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY)  
[![Security Hotspot](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/security_hotspot-16px.png 'Security Hotspot')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT)  
[![Code Smell](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/code_smell-16px.png 'Code Smell')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL) [27 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL)

[![58.1%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/50-16px.png '58.1%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_coverage&view=list) [58.1% Coverage](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_coverage&view=list)  
[![0.0%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/Duplications/3-16px.png '0.0%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_duplicated_lines_density&view=list)



![idea](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/light_bulb-16px.png 'idea') Catch issues before they fail your Quality Gate with our IDE extension ![sonarlint](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/sonarlint-16px.png 'sonarlint') [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=sonarcloud-welcome)
sonarcloud[bot]: SonarCloud Quality Gate failed.&nbsp; &nbsp; [![Quality Gate failed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/failed-16px.png 'Quality Gate failed')](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=696)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG)  
[![Vulnerability](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/vulnerability-16px.png 'Vulnerability')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY)  
[![Security Hotspot](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/security_hotspot-16px.png 'Security Hotspot')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT)  
[![Code Smell](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/code_smell-16px.png 'Code Smell')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL) [17 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL)

[![60.2%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/60-16px.png '60.2%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_coverage&view=list) [60.2% Coverage](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_coverage&view=list)  
[![0.0%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/Duplications/3-16px.png '0.0%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_duplicated_lines_density&view=list)



![idea](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/light_bulb-16px.png 'idea') Catch issues before they fail your Quality Gate with our IDE extension ![sonarlint](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/sonarlint-16px.png 'sonarlint') [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=sonarcloud-welcome)
k-wall: > Oh, also @k-wall WDYT about splitting out the KMS part of this (assuming you're happy with it) into a separate PR for merging in the short term, while we continue to hack around with the actual encryption filter part in draft PRs?

+1 to that.
sonarcloud[bot]: SonarCloud Quality Gate failed.&nbsp; &nbsp; [![Quality Gate failed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/failed-16px.png 'Quality Gate failed')](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=696)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG)  
[![Vulnerability](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/vulnerability-16px.png 'Vulnerability')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY)  
[![Security Hotspot](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/security_hotspot-16px.png 'Security Hotspot')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT)  
[![Code Smell](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/code_smell-16px.png 'Code Smell')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL) [20 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL)

[![65.2%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/60-16px.png '65.2%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_coverage&view=list) [65.2% Coverage](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_coverage&view=list)  
[![0.0%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/Duplications/3-16px.png '0.0%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_duplicated_lines_density&view=list)



![idea](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/light_bulb-16px.png 'idea') Catch issues before they fail your Quality Gate with our IDE extension ![sonarlint](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/sonarlint-16px.png 'sonarlint') [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=sonarcloud-welcome)
tombentley: Rebased following the merge of the Kms PR
sonarcloud[bot]: SonarCloud Quality Gate failed.&nbsp; &nbsp; [![Quality Gate failed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/failed-16px.png 'Quality Gate failed')](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=696)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG)  
[![Vulnerability](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/vulnerability-16px.png 'Vulnerability')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY)  
[![Security Hotspot](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/security_hotspot-16px.png 'Security Hotspot')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT)  
[![Code Smell](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/code_smell-16px.png 'Code Smell')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL) [17 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL)

[![56.1%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/50-16px.png '56.1%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_coverage&view=list) [56.1% Coverage](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_coverage&view=list)  
[![0.0%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/Duplications/3-16px.png '0.0%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_duplicated_lines_density&view=list)



![idea](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/light_bulb-16px.png 'idea') Catch issues before they fail your Quality Gate with our IDE extension ![sonarlint](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/sonarlint-16px.png 'sonarlint') [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=sonarcloud-welcome)
sonarcloud[bot]: SonarCloud Quality Gate failed.&nbsp; &nbsp; [![Quality Gate failed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/failed-16px.png 'Quality Gate failed')](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=696)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG)  
[![Vulnerability](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/vulnerability-16px.png 'Vulnerability')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY)  
[![Security Hotspot](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/security_hotspot-16px.png 'Security Hotspot')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT)  
[![Code Smell](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/code_smell-16px.png 'Code Smell')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL) [12 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL)

[![53.3%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/50-16px.png '53.3%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_coverage&view=list) [53.3% Coverage](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_coverage&view=list)  
[![0.0%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/Duplications/3-16px.png '0.0%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_duplicated_lines_density&view=list)



![idea](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/light_bulb-16px.png 'idea') Catch issues before they fail your Quality Gate with our IDE extension ![sonarlint](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/sonarlint-16px.png 'sonarlint') [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=sonarcloud-welcome)
tombentley: Rebased again
sonarcloud[bot]: Kudos, SonarCloud Quality Gate passed!&nbsp; &nbsp; [![Quality Gate passed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/passed-16px.png 'Quality Gate passed')](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=696)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG)  
[![Vulnerability](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/vulnerability-16px.png 'Vulnerability')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY)  
[![Security Hotspot](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/security_hotspot-16px.png 'Security Hotspot')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT)  
[![Code Smell](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/code_smell-16px.png 'Code Smell')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL) [9 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL)

[![81.0%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/60-16px.png '81.0%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_coverage&view=list) [81.0% Coverage](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_coverage&view=list)  
[![0.0%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/Duplications/3-16px.png '0.0%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_duplicated_lines_density&view=list)


k-wall: @tombentley did you kill EnvelopeEncryptionFilterIT intentionally?
tombentley: Finally getting back to this today. I think we need to address the following prior to merge:

* [ ] Rebase
* [ ] Restore the EncryptionIT 
* [ ] add test(s) that the caching means we're not generating DEKs for every produce request 
* [ ] add a test/assertions that headers are actually getting encrypted
* [ ] add a test that we can't use a KeyContext beyond its allowed number of encryptions or ttl. 

Other things, we could do in this PR or be deferred for a later PR, but we should do prior to a release:

* [ ] Encrypt to a single blob
* [ ] Use a single header
* [ ] Ensure all our serialized data has an explicit version, so we can guarantee compatibility. Document the schemas somewhere. 
* [ ] Change the `TopicNameBasedKekSelector` so that it returns an `EncryptionScheme`, rather than just the KEK. This will allow finer grained control over _what_ gets encrypted as well as _how_, as well as flexibility for future changes/options. 
* [ ] Add support for optional AEAD (Either globally through configuration, or finer grained via the `EncryptionScheme`; I think global is fine, because you either care about that threat vector or you don't; you're unlikely to care for topic A but not for topic B)

Is there anything I've missed in either of these lists @k-wall @SamBarker @robobario @gracegrimwood ?
robobario: > add a test/assertions that headers are actually getting encrypted
 
 Do you mean add a test at an integration level or improve an existing unit test like `InBandKeyManagerTest#shouldEncryptRecordHeaders()`?
 
 > add a test that we can't use a KeyContext beyond its allowed number of encryptions or ttl.
 
 As above, there is a test `KeyContextTest#testLifecycle` that checks context throws if you encode beyond remainingEncryptions. Are you thinking of a higher level test on `InBandKeyManager` that checks that the DEK is changed at appropriate times?
tombentley: > Do you mean add a test at an integration level

More tests are always welcome, but I was thinking IT in this case.

> Are you thinking of a higher level test on `InBandKeyManager` that checks that the DEK is changed at appropriate times?

Yes, higher level. 
tombentley: > * [ ] add test(s) that the caching means we're not generating DEKs for every produce request

I have got this locally. I'll add it once I've merged Keith's PR restoring the IT. 
robobario: Added https://github.com/tombentley/kproxy/pull/8, I think there's a problem with encrypting more than one record value in a batch because outputSize isn't called between each call to `encrypt`.

Is there an example kroxylicious config around to start up the encryption filter against a local kafka for experimenting? I see Keith's k8s example PR has some config in a config-map is that compatible with this branch @k-wall?
sonarcloud[bot]: Kudos, SonarCloud Quality Gate passed!&nbsp; &nbsp; [![Quality Gate passed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/passed-16px.png 'Quality Gate passed')](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=696)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG)  
[![Vulnerability](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/vulnerability-16px.png 'Vulnerability')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY)  
[![Security Hotspot](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/security_hotspot-16px.png 'Security Hotspot')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT)  
[![Code Smell](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/code_smell-16px.png 'Code Smell')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL) [9 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL)

[![81.0%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/60-16px.png '81.0%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_coverage&view=list) [81.0% Coverage](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_coverage&view=list)  
[![0.0%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/Duplications/3-16px.png '0.0%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_duplicated_lines_density&view=list)


sonarcloud[bot]: Kudos, SonarCloud Quality Gate passed!&nbsp; &nbsp; [![Quality Gate passed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/passed-16px.png 'Quality Gate passed')](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=696)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG)  
[![Vulnerability](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/vulnerability-16px.png 'Vulnerability')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY)  
[![Security Hotspot](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/security_hotspot-16px.png 'Security Hotspot')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT)  
[![Code Smell](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/code_smell-16px.png 'Code Smell')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL) [9 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL)

[![81.3%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/60-16px.png '81.3%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_coverage&view=list) [81.3% Coverage](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_coverage&view=list)  
[![0.0%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/Duplications/3-16px.png '0.0%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_duplicated_lines_density&view=list)


sonarcloud[bot]: Kudos, SonarCloud Quality Gate passed!&nbsp; &nbsp; [![Quality Gate passed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/passed-16px.png 'Quality Gate passed')](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=696)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG)  
[![Vulnerability](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/vulnerability-16px.png 'Vulnerability')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY)  
[![Security Hotspot](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/security_hotspot-16px.png 'Security Hotspot')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT)  
[![Code Smell](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/code_smell-16px.png 'Code Smell')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL) [9 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL)

[![81.3%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/60-16px.png '81.3%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_coverage&view=list) [81.3% Coverage](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_coverage&view=list)  
[![0.0%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/Duplications/3-16px.png '0.0%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_duplicated_lines_density&view=list)


sonarcloud[bot]: Kudos, SonarCloud Quality Gate passed!&nbsp; &nbsp; [![Quality Gate passed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/passed-16px.png 'Quality Gate passed')](https://sonarcloud.io/dashboard?id=kroxylicious_kroxylicious&pullRequest=696)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=BUG)  
[![Vulnerability](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/vulnerability-16px.png 'Vulnerability')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=VULNERABILITY)  
[![Security Hotspot](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/security_hotspot-16px.png 'Security Hotspot')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=SECURITY_HOTSPOT)  
[![Code Smell](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/code_smell-16px.png 'Code Smell')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL) [![A](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A-16px.png 'A')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL) [9 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=696&resolved=false&types=CODE_SMELL)

[![93.0%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/90-16px.png '93.0%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_coverage&view=list) [93.0% Coverage](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_coverage&view=list)  
[![0.0%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/Duplications/3-16px.png '0.0%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=696&metric=new_duplicated_lines_density&view=list)



Patch:
diff --git a/kroxylicious-bom/pom.xml b/kroxylicious-bom/pom.xml
index 7f6ef5bc..4e6609fa 100644
--- a/kroxylicious-bom/pom.xml
+++ b/kroxylicious-bom/pom.xml
@@ -123,6 +123,12 @@
                 <version>${project.version}</version>
             </dependency>
 
+            <dependency>
+                <groupId>io.kroxylicious</groupId>
+                <artifactId>kroxylicious-encryption</artifactId>
+                <version>${project.version}</version>
+            </dependency>
+
             <dependency>
                 <groupId>io.kroxylicious</groupId>
                 <artifactId>kroxylicious-multitenant</artifactId>
diff --git a/kroxylicious-filters/kroxylicious-encryption/pom.xml b/kroxylicious-filters/kroxylicious-encryption/pom.xml
new file mode 100644
index 00000000..7e982bac
--- /dev/null
+++ b/kroxylicious-filters/kroxylicious-encryption/pom.xml
@@ -0,0 +1,108 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+    Copyright Kroxylicious Authors.
+
+    Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <parent>
+        <groupId>io.kroxylicious</groupId>
+        <artifactId>kroxylicious-filter-parent</artifactId>
+        <version>0.4.0-SNAPSHOT</version>
+        <relativePath>../pom.xml</relativePath>
+    </parent>
+
+    <artifactId>kroxylicious-encryption</artifactId>
+    <packaging>jar</packaging>
+
+    <name>Kroxylicious Encryption</name>
+    <description>Kroxylicious Filter to encrypt and decrypt records
+    </description>
+
+    <dependencies>
+        <!-- project dependencies - runtime and compile -->
+        <dependency>
+            <groupId>io.kroxylicious</groupId>
+            <artifactId>kroxylicious-api</artifactId>
+        </dependency>
+        
+	    <dependency>
+            <groupId>io.kroxylicious</groupId>
+            <artifactId>kroxylicious-kms</artifactId>
+        </dependency>
+
+        <!-- third party dependencies - runtime and compile -->
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-databind</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>com.github.spotbugs</groupId>
+            <artifactId>spotbugs-annotations</artifactId>
+            <scope>provided</scope>
+        </dependency>
+
+        <!-- third party dependencies - test -->
+        <dependency>
+            <groupId>io.kroxylicious</groupId>
+            <artifactId>kroxylicious-kms-provider-kroxylicious-inmemory</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.dataformat</groupId>
+            <artifactId>jackson-dataformat-yaml</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>com.flipkart.zjsonpatch</groupId>
+            <artifactId>zjsonpatch</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>com.google.guava</groupId>
+            <artifactId>guava</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.logging.log4j</groupId>
+            <artifactId>log4j-slf4j2-impl</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.assertj</groupId>
+            <artifactId>assertj-core</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-params</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-core</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-junit-jupiter</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.hamcrest</groupId>
+            <artifactId>hamcrest</artifactId>
+        </dependency>
+    </dependencies>
+</project>
\ No newline at end of file
diff --git a/kroxylicious-filters/kroxylicious-encryption/src/main/java/io/kroxylicious/filter/encryption/EncryptionException.java b/kroxylicious-filters/kroxylicious-encryption/src/main/java/io/kroxylicious/filter/encryption/EncryptionException.java
new file mode 100644
index 00000000..cd288222
--- /dev/null
+++ b/kroxylicious-filters/kroxylicious-encryption/src/main/java/io/kroxylicious/filter/encryption/EncryptionException.java
@@ -0,0 +1,17 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.filter.encryption;
+
+public class EncryptionException extends RuntimeException {
+    public EncryptionException(Throwable e) {
+        super(e);
+    }
+
+    public EncryptionException(String message) {
+        super(message);
+    }
+}
diff --git a/kroxylicious-filters/kroxylicious-encryption/src/main/java/io/kroxylicious/filter/encryption/EncryptionScheme.java b/kroxylicious-filters/kroxylicious-encryption/src/main/java/io/kroxylicious/filter/encryption/EncryptionScheme.java
new file mode 100644
index 00000000..3cef4daf
--- /dev/null
+++ b/kroxylicious-filters/kroxylicious-encryption/src/main/java/io/kroxylicious/filter/encryption/EncryptionScheme.java
@@ -0,0 +1,28 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.filter.encryption;
+
+import java.util.Objects;
+import java.util.Set;
+
+/**
+ * Describes how a record should be encrypted
+ * @param kekId The KEK identifier to be used. Not null.
+ * @param recordFields The fields of the record that should be encrypted with the given KEK. Neither null nor empty.
+ * @param <K> The type of KEK identifier.
+ */
+public record EncryptionScheme<K>(
+                                  K kekId,
+                                  Set<RecordField> recordFields) {
+    public EncryptionScheme {
+        Objects.requireNonNull(kekId);
+        if (Objects.requireNonNull(recordFields).isEmpty()) {
+            throw new IllegalArgumentException();
+        }
+    }
+
+}
diff --git a/kroxylicious-filters/kroxylicious-encryption/src/main/java/io/kroxylicious/filter/encryption/EnvelopeEncryption.java b/kroxylicious-filters/kroxylicious-encryption/src/main/java/io/kroxylicious/filter/encryption/EnvelopeEncryption.java
new file mode 100644
index 00000000..ea5d65fd
--- /dev/null
+++ b/kroxylicious-filters/kroxylicious-encryption/src/main/java/io/kroxylicious/filter/encryption/EnvelopeEncryption.java
@@ -0,0 +1,56 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.filter.encryption;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import io.kroxylicious.filter.encryption.inband.BufferPool;
+import io.kroxylicious.filter.encryption.inband.InBandKeyManager;
+import io.kroxylicious.kms.service.Kms;
+import io.kroxylicious.kms.service.KmsService;
+import io.kroxylicious.proxy.filter.FilterFactory;
+import io.kroxylicious.proxy.filter.FilterFactoryContext;
+import io.kroxylicious.proxy.plugin.Plugin;
+import io.kroxylicious.proxy.plugin.PluginConfigurationException;
+import io.kroxylicious.proxy.plugin.PluginImplConfig;
+import io.kroxylicious.proxy.plugin.PluginImplName;
+
+import edu.umd.cs.findbugs.annotations.NonNull;
+
+/**
+ * A {@link FilterFactory} for {@link EnvelopeEncryptionFilter}.
+ */
+@Plugin(configType = EnvelopeEncryption.Config.class)
+public class EnvelopeEncryption<K, E> implements FilterFactory<EnvelopeEncryption.Config, EnvelopeEncryption.Config> {
+
+    record Config(
+                  @JsonProperty(required = true) @PluginImplName(KmsService.class) String kms,
+                  @PluginImplConfig(implNameProperty = "kms") Object kmsConfig,
+
+                  @JsonProperty(required = true) @PluginImplName(KekSelectorService.class) String selector,
+                  @PluginImplConfig(implNameProperty = "selector") Object selectorConfig) {
+
+    }
+
+    @Override
+    public Config initialize(FilterFactoryContext context, Config config) throws PluginConfigurationException {
+        return config;
+    }
+
+    @NonNull
+    @Override
+    public EnvelopeEncryptionFilter<K> createFilter(FilterFactoryContext context, Config configuration) {
+        KmsService<Object, K, E> kmsPlugin = context.pluginInstance(KmsService.class, configuration.kms());
+        Kms<K, E> kms = kmsPlugin.buildKms(configuration.kmsConfig());
+
+        var keyManager = new InBandKeyManager<>(kms, BufferPool.allocating());
+
+        KekSelectorService<Object, K> ksPlugin = context.pluginInstance(KekSelectorService.class, configuration.selector());
+        TopicNameBasedKekSelector<K> kekSelector = ksPlugin.buildSelector(kms, configuration.selectorConfig());
+        return new EnvelopeEncryptionFilter<>(keyManager, kekSelector);
+    }
+}
diff --git a/kroxylicious-filters/kroxylicious-encryption/src/main/java/io/kroxylicious/filter/encryption/EnvelopeEncryptionFilter.java b/kroxylicious-filters/kroxylicious-encryption/src/main/java/io/kroxylicious/filter/encryption/EnvelopeEncryptionFilter.java
new file mode 100644
index 00000000..4bfd48a4
--- /dev/null
+++ b/kroxylicious-filters/kroxylicious-encryption/src/main/java/io/kroxylicious/filter/encryption/EnvelopeEncryptionFilter.java
@@ -0,0 +1,181 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.filter.encryption;
+
+import java.util.ArrayList;
+import java.util.EnumSet;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import java.util.stream.StreamSupport;
+
+import org.apache.kafka.common.message.FetchResponseData;
+import org.apache.kafka.common.message.FetchResponseData.FetchableTopicResponse;
+import org.apache.kafka.common.message.FetchResponseData.PartitionData;
+import org.apache.kafka.common.message.ProduceRequestData;
+import org.apache.kafka.common.message.ProduceRequestData.TopicProduceData;
+import org.apache.kafka.common.message.RequestHeaderData;
+import org.apache.kafka.common.message.ResponseHeaderData;
+import org.apache.kafka.common.record.MemoryRecords;
+import org.apache.kafka.common.record.MemoryRecordsBuilder;
+import org.apache.kafka.common.record.RecordBatch;
+import org.apache.kafka.common.utils.ByteBufferOutputStream;
+import org.slf4j.Logger;
+
+import io.kroxylicious.proxy.filter.FetchResponseFilter;
+import io.kroxylicious.proxy.filter.FilterContext;
+import io.kroxylicious.proxy.filter.ProduceRequestFilter;
+import io.kroxylicious.proxy.filter.RequestFilterResult;
+import io.kroxylicious.proxy.filter.ResponseFilterResult;
+
+import edu.umd.cs.findbugs.annotations.NonNull;
+
+import static org.slf4j.LoggerFactory.getLogger;
+
+/**
+ * A filter for encrypting and decrypting records using envelope encryption
+ * @param <K> The type of KEK reference
+ */
+public class EnvelopeEncryptionFilter<K>
+        implements ProduceRequestFilter, FetchResponseFilter {
+    private static final Logger log = getLogger(EnvelopeEncryptionFilter.class);
+    private final TopicNameBasedKekSelector<K> kekSelector;
+
+    private final KeyManager<K> keyManager;
+
+    EnvelopeEncryptionFilter(KeyManager<K> keyManager, TopicNameBasedKekSelector<K> kekSelector) {
+        this.kekSelector = kekSelector;
+        this.keyManager = keyManager;
+    }
+
+    @SuppressWarnings("unchecked")
+    public static <T> CompletionStage<List<T>> join(List<? extends CompletionStage<T>> stages) {
+        CompletableFuture<T>[] futures = new CompletableFuture[stages.size()];
+        for (int i = 0; i < stages.size(); i++) {
+            futures[i] = stages.get(i).toCompletableFuture();
+        }
+        return CompletableFuture.allOf(futures)
+                .thenApply(ignored -> Stream.of(futures).map(CompletableFuture::join).toList());
+    }
+
+    @Override
+    public CompletionStage<RequestFilterResult> onProduceRequest(short apiVersion,
+                                                                 RequestHeaderData header,
+                                                                 ProduceRequestData request,
+                                                                 FilterContext context) {
+        return maybeEncodeProduce(request, context)
+                .thenCompose(yy -> context.forwardRequest(header, request));
+    }
+
+    private CompletionStage<ProduceRequestData> maybeEncodeProduce(ProduceRequestData request, FilterContext context) {
+        var topicNameToData = request.topicData().stream().collect(Collectors.toMap(TopicProduceData::name, Function.identity()));
+        return kekSelector.selectKek(topicNameToData.keySet()) // figure out what keks we need
+                .thenCompose(kekMap -> {
+                    var futures = kekMap.entrySet().stream().flatMap(e -> {
+                        String topicName = e.getKey();
+                        var kekId = e.getValue();
+                        TopicProduceData tpd = topicNameToData.get(topicName);
+                        return tpd.partitionData().stream().map(ppd -> {
+                            // handle case where this topic is to be left unencrypted
+                            if (kekId == null) {
+                                return CompletableFuture.completedStage(ppd);
+                            }
+                            MemoryRecords records = (MemoryRecords) ppd.records();
+                            MemoryRecordsBuilder builder = recordsBuilder(allocateBufferForEncode(records, context), records);
+                            var encryptionRequests = recordStream(records).toList();
+                            return keyManager.encrypt(
+                                    new EncryptionScheme<>(kekId, EnumSet.of(RecordField.RECORD_VALUE)),
+                                    encryptionRequests,
+                                    (kafkaRecord, encryptedValue, headers) -> builder.append(kafkaRecord.timestamp(), kafkaRecord.key(), encryptedValue, headers))
+                                    .thenApply(ignored -> ppd.setRecords(builder.build()));
+                        });
+                    }).toList();
+                    return join(futures).thenApply(x -> request);
+                }).exceptionallyCompose(throwable -> {
+                    log.warn("failed to encrypt records", throwable);
+                    return CompletableFuture.failedStage(throwable);
+                });
+    }
+
+    private ByteBufferOutputStream allocateBufferForEncode(MemoryRecords records, FilterContext context) {
+        int sizeEstimate = 2 * records.sizeInBytes();
+        // Accurate estimation is tricky without knowing the record sizes
+        return context.createByteBufferOutputStream(sizeEstimate);
+    }
+
+    @Override
+    public CompletionStage<ResponseFilterResult> onFetchResponse(short apiVersion, ResponseHeaderData header, FetchResponseData response, FilterContext context) {
+        return maybeDecodeFetch(response.responses(), context)
+                .thenCompose(responses -> context.forwardResponse(header, response.setResponses(responses)))
+                .exceptionallyCompose(throwable -> {
+                    log.warn("failed to decrypt records", throwable);
+                    return CompletableFuture.failedStage(throwable);
+                });
+    }
+
+    private CompletionStage<List<FetchableTopicResponse>> maybeDecodeFetch(List<FetchableTopicResponse> topics, FilterContext context) {
+        List<CompletionStage<FetchableTopicResponse>> result = new ArrayList<>(topics.size());
+        for (FetchableTopicResponse topicData : topics) {
+            result.add(maybeDecodePartitions(topicData.partitions(), context).thenApply(kk -> {
+                topicData.setPartitions(kk);
+                return topicData;
+            }));
+        }
+        return join(result);
+    }
+
+    private CompletionStage<List<PartitionData>> maybeDecodePartitions(List<PartitionData> partitions, FilterContext context) {
+        List<CompletionStage<PartitionData>> result = new ArrayList<>(partitions.size());
+        for (PartitionData partitionData : partitions) {
+            if (!(partitionData.records() instanceof MemoryRecords)) {
+                throw new IllegalStateException();
+            }
+            result.add(maybeDecodeRecords(partitionData, (MemoryRecords) partitionData.records(), context));
+        }
+        return join(result);
+    }
+
+    private CompletionStage<PartitionData> maybeDecodeRecords(PartitionData fpr,
+                                                              MemoryRecords memoryRecords,
+                                                              FilterContext context) {
+        MemoryRecordsBuilder builder = recordsBuilder(allocateBufferForDecode(memoryRecords, context), memoryRecords);
+        return keyManager.decrypt(recordStream(memoryRecords).toList(),
+                (kafkaRecord, plaintextBuffer, headers) -> builder.append(kafkaRecord.timestamp(), kafkaRecord.key(), plaintextBuffer, headers))
+                .thenApply(ignored -> builder.build())
+                .thenApply(fpr::setRecords);
+    }
+
+    private ByteBufferOutputStream allocateBufferForDecode(MemoryRecords memoryRecords, FilterContext context) {
+        int sizeEstimate = memoryRecords.sizeInBytes();
+        return context.createByteBufferOutputStream(sizeEstimate);
+    }
+
+    @NonNull
+    private static Stream<org.apache.kafka.common.record.Record> recordStream(MemoryRecords memoryRecords) {
+        return StreamSupport.stream(memoryRecords.records().spliterator(), false);
+    }
+
+    private static MemoryRecordsBuilder recordsBuilder(@NonNull ByteBufferOutputStream buffer, @NonNull MemoryRecords records) {
+        RecordBatch firstBatch = records.firstBatch();
+        return new MemoryRecordsBuilder(buffer,
+                firstBatch.magic(),
+                firstBatch.compressionType(), // TODO we might not want to use the client's compression
+                firstBatch.timestampType(),
+                firstBatch.baseOffset(),
+                0L,
+                firstBatch.producerId(),
+                firstBatch.producerEpoch(),
+                firstBatch.baseSequence(),
+                firstBatch.isTransactional(),
+                firstBatch.isControlBatch(),
+                firstBatch.partitionLeaderEpoch(),
+                0);
+    }
+}
\ No newline at end of file
diff --git a/kroxylicious-filters/kroxylicious-encryption/src/main/java/io/kroxylicious/filter/encryption/KekSelectorService.java b/kroxylicious-filters/kroxylicious-encryption/src/main/java/io/kroxylicious/filter/encryption/KekSelectorService.java
new file mode 100644
index 00000000..39372529
--- /dev/null
+++ b/kroxylicious-filters/kroxylicious-encryption/src/main/java/io/kroxylicious/filter/encryption/KekSelectorService.java
@@ -0,0 +1,22 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.filter.encryption;
+
+import io.kroxylicious.kms.service.Kms;
+
+import edu.umd.cs.findbugs.annotations.NonNull;
+
+/**
+ * KekSelectorService
+ * @param <C> The config type
+ * @param <K> the type of key
+ */
+public interface KekSelectorService<C, K> {
+    @NonNull
+    TopicNameBasedKekSelector<K> buildSelector(@NonNull Kms<K, ?> kms, C options);
+
+}
diff --git a/kroxylicious-filters/kroxylicious-encryption/src/main/java/io/kroxylicious/filter/encryption/KeyManager.java b/kroxylicious-filters/kroxylicious-encryption/src/main/java/io/kroxylicious/filter/encryption/KeyManager.java
new file mode 100644
index 00000000..f54ee0ac
--- /dev/null
+++ b/kroxylicious-filters/kroxylicious-encryption/src/main/java/io/kroxylicious/filter/encryption/KeyManager.java
@@ -0,0 +1,44 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.filter.encryption;
+
+import java.util.List;
+import java.util.concurrent.CompletionStage;
+
+import org.apache.kafka.common.record.Record;
+
+import edu.umd.cs.findbugs.annotations.NonNull;
+
+/**
+ * A manager of (data) encryption keys supporting encryption and decryption operations,
+ * encapsulating access to the data encryption keys.
+ * @param <K> The type of KEK id.
+ */
+public interface KeyManager<K> {
+
+    /**
+     * Asynchronously encrypt the given {@code recordRequests} using the current DEK for the given KEK, calling the given receiver for each encrypted record
+     * @param encryptionScheme The encryption scheme.
+     * @param records The requests to be encrypted.
+     * @param receiver The receiver of the encrypted buffers.
+     * @return A completion stage that completes when all the records in the given {@code recordRequests} have been processed.
+     */
+    @NonNull
+    CompletionStage<Void> encrypt(@NonNull EncryptionScheme<K> encryptionScheme,
+                                  @NonNull List<? extends Record> records,
+                                  @NonNull Receiver receiver);
+
+    /**
+     * Asynchronously decrypt the given {@code kafkaRecords} (if they were, in fact, encrypted), calling the given {@code receiver} with the plaintext
+     * @param records The records
+     * @param receiver The receiver of the plaintext buffers
+     * @return A completion stage that completes when the record has been processed.
+     */
+    @NonNull
+    CompletionStage<Void> decrypt(@NonNull List<? extends Record> records,
+                                  @NonNull Receiver receiver);
+}
diff --git a/kroxylicious-filters/kroxylicious-encryption/src/main/java/io/kroxylicious/filter/encryption/Receiver.java b/kroxylicious-filters/kroxylicious-encryption/src/main/java/io/kroxylicious/filter/encryption/Receiver.java
new file mode 100644
index 00000000..62ff4b90
--- /dev/null
+++ b/kroxylicious-filters/kroxylicious-encryption/src/main/java/io/kroxylicious/filter/encryption/Receiver.java
@@ -0,0 +1,25 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.filter.encryption;
+
+import java.nio.ByteBuffer;
+
+import org.apache.kafka.common.header.Header;
+import org.apache.kafka.common.record.Record;
+
+/**
+ * Something that receives the result of an encryption or decryption operation
+ */
+public interface Receiver {
+    /**
+     * Receive the ciphertext (encryption) or the plaintext (decryption) associated with the given record..
+     *
+     * @param kafkaRecord The record on which to base the revised record
+     * @param value The ciphertext or plaintext buffer.
+     */
+    void accept(Record kafkaRecord, ByteBuffer value, Header[] headers);
+}
diff --git a/kroxylicious-filters/kroxylicious-encryption/src/main/java/io/kroxylicious/filter/encryption/RecordField.java b/kroxylicious-filters/kroxylicious-encryption/src/main/java/io/kroxylicious/filter/encryption/RecordField.java
new file mode 100644
index 00000000..5fceb6c0
--- /dev/null
+++ b/kroxylicious-filters/kroxylicious-encryption/src/main/java/io/kroxylicious/filter/encryption/RecordField.java
@@ -0,0 +1,43 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.filter.encryption;
+
+import java.util.EnumSet;
+import java.util.Set;
+
+/**
+ * Enumerates the parts of a Kafka Record that can be encrypted.
+ */
+public enum RecordField {
+    RECORD_VALUE((byte) 1),
+    RECORD_HEADER_VALUES((byte) (1 << 1));
+
+    private final byte code;
+
+    // RECORD_KEY((byte) (1 << 2)),
+    // RECORD_TIMESTAMP((byte) (1 << 3)
+    RecordField(byte code) {
+        this.code = code;
+    }
+
+    public static byte toBits(Set<RecordField> recordField) {
+        return (byte) recordField.stream()
+                .mapToInt(w -> w.code).reduce((x, y) -> x | y)
+                .orElse(0);
+    }
+
+    public static Set<RecordField> fromBits(byte b) {
+        var result = EnumSet.noneOf(RecordField.class);
+        if ((b & RecordField.RECORD_VALUE.code) != 0) {
+            result.add(RecordField.RECORD_VALUE);
+        }
+        if ((b & RecordField.RECORD_HEADER_VALUES.code) != 0) {
+            result.add(RecordField.RECORD_HEADER_VALUES);
+        }
+        return result;
+    }
+}
diff --git a/kroxylicious-filters/kroxylicious-encryption/src/main/java/io/kroxylicious/filter/encryption/TemplateKekSelector.java b/kroxylicious-filters/kroxylicious-encryption/src/main/java/io/kroxylicious/filter/encryption/TemplateKekSelector.java
new file mode 100644
index 00000000..9685c814
--- /dev/null
+++ b/kroxylicious-filters/kroxylicious-encryption/src/main/java/io/kroxylicious/filter/encryption/TemplateKekSelector.java
@@ -0,0 +1,97 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.filter.encryption;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+import java.util.regex.Pattern;
+
+import io.kroxylicious.kms.service.Kms;
+import io.kroxylicious.kms.service.UnknownAliasException;
+import io.kroxylicious.proxy.plugin.Plugin;
+
+import edu.umd.cs.findbugs.annotations.NonNull;
+
+@Plugin(configType = TemplateKekSelector.Config.class)
+public class TemplateKekSelector<K> implements KekSelectorService<TemplateKekSelector.Config, K> {
+    public record Config(String template) {}
+
+    @NonNull
+    @Override
+    public TopicNameBasedKekSelector<K> buildSelector(@NonNull Kms<K, ?> kms, Config config) {
+        return new KekSelector<>(kms, config.template());
+    }
+
+    static class KekSelector<K> extends TopicNameBasedKekSelector<K> {
+
+        public static final Pattern PATTERN = Pattern.compile("\\$\\{(.*?)}");
+        private final String template;
+        private final Kms<K, ?> kms;
+
+        KekSelector(@NonNull Kms<K, ?> kms, @NonNull String template) {
+            var matcher = PATTERN.matcher(Objects.requireNonNull(template));
+            while (matcher.find()) {
+                if (matcher.group(1).equals("topicName")) {
+                    continue;
+                }
+                throw new IllegalArgumentException("Unknown template parameter: " + matcher.group(1));
+            }
+            this.template = Objects.requireNonNull(template);
+            this.kms = Objects.requireNonNull(kms);
+        }
+
+        private record Pair<K>(String topicName, K kekId) {}
+
+        @NonNull
+        @Override
+        public CompletionStage<Map<String, K>> selectKek(@NonNull Set<String> topicNames) {
+            var collect = topicNames.stream()
+                    .map(
+                            topicName -> kms.resolveAlias(evaluateTemplate(topicName))
+                                    .exceptionallyCompose(e -> {
+                                        if (e instanceof UnknownAliasException) {
+                                            return CompletableFuture.completedFuture(null);
+                                        }
+                                        else {
+                                            return CompletableFuture.failedFuture(e);
+                                        }
+                                    })
+                                    .thenApply(kekId -> new Pair<>(topicName, kekId)))
+                    .toList();
+            return EnvelopeEncryptionFilter.join(collect).thenApply(list -> {
+                // Note we can't use `java.util.stream...(Collectors.toMap())` to build the map, because it has null values
+                // which Collectors.toMap() does now allow.
+                Map<String, K> map = new HashMap<>();
+                for (Pair<K> pair : list) {
+                    map.put(pair.topicName(), pair.kekId());
+                }
+                return map;
+            });
+        }
+
+        String evaluateTemplate(String topicName) {
+            var matcher = PATTERN.matcher(template);
+            StringBuilder sb = new StringBuilder();
+            while (matcher.find()) {
+                String replacement;
+                if (matcher.group(1).equals("topicName")) {
+                    replacement = topicName;
+                }
+                else { // this should be impossible because of the check in the constructor
+                    throw new IllegalStateException();
+                }
+                matcher.appendReplacement(sb, replacement);
+            }
+            matcher.appendTail(sb);
+            return sb.toString();
+        }
+    }
+}
diff --git a/kroxylicious-filters/kroxylicious-encryption/src/main/java/io/kroxylicious/filter/encryption/TopicNameBasedKekSelector.java b/kroxylicious-filters/kroxylicious-encryption/src/main/java/io/kroxylicious/filter/encryption/TopicNameBasedKekSelector.java
new file mode 100644
index 00000000..12e3a4cc
--- /dev/null
+++ b/kroxylicious-filters/kroxylicious-encryption/src/main/java/io/kroxylicious/filter/encryption/TopicNameBasedKekSelector.java
@@ -0,0 +1,28 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.filter.encryption;
+
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletionStage;
+
+import edu.umd.cs.findbugs.annotations.NonNull;
+
+/**
+ * KEK selection based on topic name
+ * @param <K> the type of key
+ */
+public abstract class TopicNameBasedKekSelector<K> {
+
+    /**
+     * Returns a completion stage whose value, on successful completion, is a map from each of the given topic
+     * names to the KEK id to use for encrypting records in that topic.
+     * @param topicNames A set of topic names
+     * @return A completion stage for the map form topic name to KEK id.
+     */
+    public abstract @NonNull CompletionStage<Map<String, K>> selectKek(@NonNull Set<String> topicNames);
+}
diff --git a/kroxylicious-filters/kroxylicious-encryption/src/main/java/io/kroxylicious/filter/encryption/inband/AesGcmEncryptor.java b/kroxylicious-filters/kroxylicious-encryption/src/main/java/io/kroxylicious/filter/encryption/inband/AesGcmEncryptor.java
new file mode 100644
index 00000000..083fa744
--- /dev/null
+++ b/kroxylicious-filters/kroxylicious-encryption/src/main/java/io/kroxylicious/filter/encryption/inband/AesGcmEncryptor.java
@@ -0,0 +1,176 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.filter.encryption.inband;
+
+import java.nio.ByteBuffer;
+import java.security.GeneralSecurityException;
+import java.util.Arrays;
+import java.util.Objects;
+
+import javax.annotation.concurrent.NotThreadSafe;
+import javax.crypto.Cipher;
+import javax.crypto.SecretKey;
+import javax.crypto.spec.GCMParameterSpec;
+import javax.security.auth.DestroyFailedException;
+import javax.security.auth.Destroyable;
+
+import io.kroxylicious.filter.encryption.EncryptionException;
+
+import edu.umd.cs.findbugs.annotations.NonNull;
+
+@NotThreadSafe
+class AesGcmEncryptor implements Destroyable {
+
+    private int lastCall;
+    private static final int OUTPUT_SIZE = 0;
+    private static final int ENCRYPT = 1;
+    private static final int DECRYPT = 2;
+
+    private SecretKey key;
+    private Cipher cipher;
+    private final int numAuthBits;
+    private final byte[] iv;
+    private final AesGcmIvGenerator ivGenerator;
+
+    private AesGcmEncryptor(int ivSizeBytes, AesGcmIvGenerator ivGenerator, @NonNull SecretKey key) {
+        // NIST SP.800-38D recommends 96 bit for recommendation about the iv length and generation
+        this.iv = new byte[ivSizeBytes];
+        this.ivGenerator = ivGenerator;
+        this.numAuthBits = 128;
+        this.key = Objects.requireNonNull(key);
+        try {
+            this.cipher = Cipher.getInstance("AES_256/GCM/NoPadding");
+        }
+        catch (GeneralSecurityException e) {
+            throw new EncryptionException(e);
+        }
+    }
+
+    public static AesGcmEncryptor forEncrypt(@NonNull AesGcmIvGenerator ivGenerator, @NonNull SecretKey key) {
+        Objects.requireNonNull(ivGenerator);
+        var result = new AesGcmEncryptor(ivGenerator.sizeBytes(), ivGenerator, key);
+        result.lastCall = ENCRYPT;
+        return result;
+    }
+
+    @NonNull
+    public static AesGcmEncryptor forDecrypt(@NonNull SecretKey key) {
+        var result = new AesGcmEncryptor(12, null, key);
+        result.init(Cipher.DECRYPT_MODE);
+        result.lastCall = DECRYPT;
+        return result;
+
+    }
+
+    public int outputSize(int plaintextSize) {
+        checkNotDestroyed();
+        ivGenerator.generateIv(iv);
+        init(Cipher.ENCRYPT_MODE);
+        int result = headerSize() // iv
+                + this.cipher.getOutputSize(plaintextSize);
+        lastCall = OUTPUT_SIZE;
+        return result;
+    }
+
+    private int headerSize() {
+        return Byte.BYTES // version
+                + Byte.BYTES // iv length
+                + ivGenerator.sizeBytes();
+    }
+
+    /**
+     * Encrypt the given plaintext, writing the ciphertext and any necessary extra data to the given {@code output}.
+     * @param plaintext The plaintext to encrypt.
+     * @param output The output buffer.
+     */
+    public void encrypt(@NonNull ByteBuffer plaintext, @NonNull ByteBuffer output) {
+        checkNotDestroyed();
+        if (lastCall != OUTPUT_SIZE) {
+            throw new IllegalStateException("Call to encrypt() without last call being to outputSize()");
+        }
+        lastCall = ENCRYPT;
+        int p0 = output.position();
+        output.position(p0 + headerSize());
+        try {
+            this.cipher.doFinal(plaintext, output);
+        }
+        catch (GeneralSecurityException e) {
+            throw new EncryptionException(e);
+        }
+        int p1 = output.position();
+        output.position(p0);
+        byte version = 0;
+        output.put(version);
+        output.put((byte) iv.length);
+        output.put(iv); // the iv
+        output.position(p1);
+    }
+
+    private void init(int encryptMode) {
+        var spec = new GCMParameterSpec(numAuthBits, iv);
+        try {
+            this.cipher.init(encryptMode, key, spec);
+        }
+        catch (GeneralSecurityException e) {
+            throw new EncryptionException(e);
+        }
+    }
+
+    public void decrypt(@NonNull ByteBuffer input, @NonNull ByteBuffer plaintext) {
+        checkNotDestroyed();
+        if (lastCall != DECRYPT) {
+            throw new IllegalStateException();
+        }
+        var version = input.get();
+        if (version == 0) {
+            int ivLength = input.get();
+            if (ivLength != iv.length) {
+                throw new EncryptionException("Unexpected IV length");
+            }
+            input.get(iv, 0, iv.length);
+            init(Cipher.DECRYPT_MODE);
+            try {
+                this.cipher.doFinal(input, plaintext);
+            }
+            catch (GeneralSecurityException e) {
+                throw new EncryptionException(e);
+            }
+        }
+        else {
+            throw new EncryptionException("Unknown version " + version);
+        }
+    }
+
+    private void checkNotDestroyed() {
+        if (key == null) {
+            throw new IllegalStateException();
+        }
+    }
+
+    @Override
+    public void destroy() throws DestroyFailedException {
+        ivGenerator.destroy();
+        Arrays.fill(iv, (byte) 0);
+        try {
+            if (key != null) {
+                key.destroy();
+            }
+        }
+        catch (DestroyFailedException e) {
+            throw new DestroyFailedException("On key of " + key.getClass());
+        }
+        finally {
+            key = null;
+            cipher = null;
+        }
+    }
+
+    @Override
+    public boolean isDestroyed() {
+        return key == null;
+    }
+}
diff --git a/kroxylicious-filters/kroxylicious-encryption/src/main/java/io/kroxylicious/filter/encryption/inband/AesGcmIvGenerator.java b/kroxylicious-filters/kroxylicious-encryption/src/main/java/io/kroxylicious/filter/encryption/inband/AesGcmIvGenerator.java
new file mode 100644
index 00000000..041c44b5
--- /dev/null
+++ b/kroxylicious-filters/kroxylicious-encryption/src/main/java/io/kroxylicious/filter/encryption/inband/AesGcmIvGenerator.java
@@ -0,0 +1,91 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.filter.encryption.inband;
+
+import java.security.SecureRandom;
+
+import javax.annotation.concurrent.NotThreadSafe;
+import javax.security.auth.Destroyable;
+
+import edu.umd.cs.findbugs.annotations.NonNull;
+
+/**
+ * RBG-based construction of the Initialization Vector for AES-GCM.
+ * The random field is 96 bits and the free field is empty.
+ *
+ * @see "§8.2.2 of NIST SP.800-38D: Recommendation for Block
+ * Cipher Modes of Operation:
+ * Galois/Counter Mode (GCM)
+ * and GMAC"
+ */
+@NotThreadSafe
+class AesGcmIvGenerator implements Destroyable {
+
+    private int low;
+    private int mid;
+    private int hi;
+    private boolean destroyed = false;
+
+    AesGcmIvGenerator(@NonNull SecureRandom rng) {
+        low = rng.nextInt();
+        mid = rng.nextInt();
+        hi = rng.nextInt();
+    }
+
+    int sizeBytes() {
+        return 12;
+    }
+
+    void generateIv(byte[] iv) {
+        if (destroyed) {
+            throw new IllegalStateException();
+        }
+        iv[0] = (byte) (hi >> 24);
+        iv[1] = (byte) (hi >> 16);
+        iv[2] = (byte) (hi >> 8);
+        iv[3] = (byte) (hi);
+        iv[4] = (byte) (mid >> 24);
+        iv[5] = (byte) (mid >> 16);
+        iv[6] = (byte) (mid >> 8);
+        iv[7] = (byte) (mid);
+        iv[8] = (byte) (low >> 24);
+        iv[9] = (byte) (low >> 16);
+        iv[10] = (byte) (low >> 8);
+        iv[11] = (byte) (low);
+        try {
+            low = Math.addExact(low, 1);
+        }
+        catch (ArithmeticException e) {
+            low = Integer.MIN_VALUE;
+            try {
+                mid = Math.addExact(mid, 1);
+            }
+            catch (ArithmeticException e2) {
+                mid = Integer.MIN_VALUE;
+                try {
+                    hi = Math.addExact(hi, 1);
+                }
+                catch (ArithmeticException e3) {
+                    hi = Integer.MIN_VALUE;
+                }
+            }
+        }
+    }
+
+    @Override
+    public void destroy() {
+        low = 0;
+        mid = 0;
+        hi = 0;
+        destroyed = true;
+    }
+
+    @Override
+    public boolean isDestroyed() {
+        return destroyed;
+    }
+}
diff --git a/kroxylicious-filters/kroxylicious-encryption/src/main/java/io/kroxylicious/filter/encryption/inband/BufferPool.java b/kroxylicious-filters/kroxylicious-encryption/src/main/java/io/kroxylicious/filter/encryption/inband/BufferPool.java
new file mode 100644
index 00000000..0c857fc1
--- /dev/null
+++ b/kroxylicious-filters/kroxylicious-encryption/src/main/java/io/kroxylicious/filter/encryption/inband/BufferPool.java
@@ -0,0 +1,30 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.filter.encryption.inband;
+
+import java.nio.ByteBuffer;
+
+public interface BufferPool {
+
+    ByteBuffer acquire(int size);
+
+    void release(ByteBuffer buffer);
+
+    static BufferPool allocating() {
+        return new BufferPool() {
+            @Override
+            public ByteBuffer acquire(int size) {
+                return ByteBuffer.allocate(size);
+            }
+
+            @Override
+            public void release(ByteBuffer buffer) {
+                // this impl doesn't pool, so leave it to the GC to tidy up
+            }
+        };
+    }
+}
diff --git a/kroxylicious-filters/kroxylicious-encryption/src/main/java/io/kroxylicious/filter/encryption/inband/ExhaustedDekException.java b/kroxylicious-filters/kroxylicious-encryption/src/main/java/io/kroxylicious/filter/encryption/inband/ExhaustedDekException.java
new file mode 100644
index 00000000..4528d7e7
--- /dev/null
+++ b/kroxylicious-filters/kroxylicious-encryption/src/main/java/io/kroxylicious/filter/encryption/inband/ExhaustedDekException.java
@@ -0,0 +1,15 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.filter.encryption.inband;
+
+import io.kroxylicious.filter.encryption.EncryptionException;
+
+public class ExhaustedDekException extends EncryptionException {
+    public ExhaustedDekException(String msg) {
+        super(msg);
+    }
+}
diff --git a/kroxylicious-filters/kroxylicious-encryption/src/main/java/io/kroxylicious/filter/encryption/inband/InBandKeyManager.java b/kroxylicious-filters/kroxylicious-encryption/src/main/java/io/kroxylicious/filter/encryption/inband/InBandKeyManager.java
new file mode 100644
index 00000000..3dbb39ea
--- /dev/null
+++ b/kroxylicious-filters/kroxylicious-encryption/src/main/java/io/kroxylicious/filter/encryption/inband/InBandKeyManager.java
@@ -0,0 +1,387 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.filter.encryption.inband;
+
+import java.nio.ByteBuffer;
+import java.security.SecureRandom;
+import java.util.ArrayList;
+import java.util.EnumSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Supplier;
+
+import javax.security.auth.DestroyFailedException;
+import javax.security.auth.Destroyable;
+
+import org.apache.kafka.common.header.Header;
+import org.apache.kafka.common.header.internals.RecordHeader;
+import org.apache.kafka.common.record.Record;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.kroxylicious.filter.encryption.EncryptionScheme;
+import io.kroxylicious.filter.encryption.KeyManager;
+import io.kroxylicious.filter.encryption.Receiver;
+import io.kroxylicious.filter.encryption.RecordField;
+import io.kroxylicious.kms.service.Kms;
+import io.kroxylicious.kms.service.Serde;
+
+import edu.umd.cs.findbugs.annotations.NonNull;
+import edu.umd.cs.findbugs.annotations.Nullable;
+
+/**
+ * An implementation of {@link KeyManager} that uses envelope encryption, AES-GCM and stores the KEK id and encrypted DEK
+ * alongside the record ("in-band").
+ * @param <K> The type of KEK id.
+ * @param <E> The type of the encrypted DEK.
+ */
+public class InBandKeyManager<K, E> implements KeyManager<K> {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(InBandKeyManager.class);
+    private static final Map<Class<? extends Destroyable>, Boolean> LOGGED_DESTROY_FAILED = new ConcurrentHashMap<>();
+
+    private static final String FIELDS_HEADER_NAME = "kroxylicious.io/encrypted";
+    private static final String DEK_HEADER_NAME = "kroxylicious.io/dek";
+
+    private final Kms<K, E> kms;
+    private final BufferPool bufferPool;
+    private final Serde<K> kekIdSerde;
+    private final Serde<E> edekSerde;
+    // TODO cache expiry, with key descruction
+    private final ConcurrentHashMap<K, CompletionStage<KeyContext>> keyContextCache;
+    private final ConcurrentHashMap<RecordHeader, CompletionStage<AesGcmEncryptor>> decryptorCache;
+    private final long dekTtlNanos;
+    private final int maxEncryptionsPerDek;
+
+    public InBandKeyManager(Kms<K, E> kms,
+                            BufferPool bufferPool) {
+        this.kms = kms;
+        this.bufferPool = bufferPool;
+        this.edekSerde = kms.edekSerde();
+        this.kekIdSerde = kms.keyIdSerde();
+        this.dekTtlNanos = 5_000_000_000L;
+        this.maxEncryptionsPerDek = 500_000;
+        // TODO This ^^ must be > the maximum size of a batch to avoid an infinite loop
+        this.keyContextCache = new ConcurrentHashMap<>();
+        this.decryptorCache = new ConcurrentHashMap<>();
+    }
+
+    private CompletionStage<KeyContext> getKeyContext(K key,
+                                                      Supplier<CompletionStage<KeyContext>> valueSupplier) {
+        return keyContextCache.compute(key, (k, v) -> {
+            if (v == null) {
+                return valueSupplier.get();
+                // TODO what happens if the CS complete exceptionally
+                // TODO what happens if the CS doesn't complete at all in a reasonably time frame?
+            }
+            else {
+                return v;
+            }
+        });
+    }
+
+    private CompletionStage<KeyContext> currentDekContext(@NonNull K kekId, int numRecords) {
+        return getKeyContext(kekId, makeKeyContext(kekId))
+                .thenCompose(cachedContext -> {
+                    if (!cachedContext.isExpiredForEncryption(System.nanoTime())
+                            && cachedContext.hasAtLeastRemainingEncryptions(numRecords)) {
+                        // TODO pre-populate decryptorCache?
+                        return CompletableFuture.completedFuture(cachedContext);
+                    }
+                    else {
+                        destroy(cachedContext);
+                        return currentDekContext(kekId, numRecords);
+                    }
+                });
+    }
+
+    static void destroy(Destroyable destroyable) {
+        try {
+            destroyable.destroy();
+        }
+        catch (DestroyFailedException e) {
+            var cls = destroyable.getClass();
+            LOGGED_DESTROY_FAILED.compute(cls, (k, logged) -> {
+                if (logged == null) {
+                    LOGGER.warn("Failed to destroy an instance of {}. "
+                            + "Note: this message is logged once per class even though there may be many occurrences of this event. "
+                            + "This event can happen because the JRE's SecretKeySpec class does not override destroy().",
+                            cls, e);
+                }
+                return Boolean.TRUE;
+            });
+        }
+    }
+
+    private Supplier<CompletionStage<KeyContext>> makeKeyContext(@NonNull K kekId) {
+        return () -> kms.generateDekPair(kekId)
+                .thenApply(dekPair -> {
+                    E edek = dekPair.edek();
+                    short kekIdSize = (short) kekIdSerde.sizeOf(kekId);
+                    short edekSize = (short) edekSerde.sizeOf(edek);
+                    ByteBuffer prefix = bufferPool.acquire(
+                            Short.BYTES + // kekId size
+                                    kekIdSize + // the kekId
+                                    Short.BYTES + // DEK size
+                                    edekSize); // the DEK
+                    prefix.putShort(kekIdSize);
+                    kekIdSerde.serialize(kekId, prefix);
+                    prefix.putShort(edekSize);
+                    edekSerde.serialize(edek, prefix);
+                    prefix.flip();
+
+                    return new KeyContext(prefix,
+                            System.nanoTime() + dekTtlNanos,
+                            maxEncryptionsPerDek,
+                            // Either we have a different Aes encryptor for each thread
+                            // or we need mutex
+                            // or we externalize the state
+                            AesGcmEncryptor.forEncrypt(new AesGcmIvGenerator(new SecureRandom()), dekPair.dek()));
+                });
+    }
+
+    @NonNull
+    @Override
+    @SuppressWarnings("java:S2445")
+    public CompletionStage<Void> encrypt(@NonNull EncryptionScheme<K> encryptionScheme,
+                                         @NonNull List<? extends Record> records,
+                                         @NonNull Receiver receiver) {
+
+        var fieldsHeader = createEncryptedFieldsHeader(encryptionScheme.recordFields());
+        return currentDekContext(encryptionScheme.kekId(), records.size()).thenAccept(keyContext -> {
+            var dekHeader = createEdekHeader(keyContext);
+            var maxValuePlaintextSize = encryptionScheme.recordFields().contains(RecordField.RECORD_VALUE)
+                    ? records.stream().mapToInt(Record::valueSize).max().orElse(-1)
+                    : -1;
+            synchronized (keyContext) {
+                var valueCiphertext = maxValuePlaintextSize >= 0 ? bufferPool.acquire(keyContext.encodedSize(maxValuePlaintextSize)) : null;
+                try {
+                    encryptRecords(encryptionScheme, keyContext, records, fieldsHeader, dekHeader, valueCiphertext, receiver);
+                }
+                finally {
+                    if (maxValuePlaintextSize >= 0) {
+                        bufferPool.release(valueCiphertext);
+                    }
+                }
+            }
+        });
+    }
+
+    private void encryptRecords(@NonNull EncryptionScheme<K> encryptionScheme,
+                                @NonNull KeyContext keyContext,
+                                @NonNull List<? extends Record> records,
+                                @NonNull RecordHeader fieldsHeader,
+                                @NonNull RecordHeader dekHeader,
+                                ByteBuffer valueCiphertext,
+                                @NonNull Receiver receiver) {
+        records.forEach(kafkaRecord -> {
+            ByteBuffer transformedValue;
+            Header[] headers = kafkaRecord.headers();
+            if (encryptionScheme.recordFields().contains(RecordField.RECORD_VALUE)) {
+                transformedValue = encryptRecordValue(keyContext, kafkaRecord, valueCiphertext);
+            }
+            else {
+                transformedValue = kafkaRecord.value();
+            }
+            if (encryptionScheme.recordFields().contains(RecordField.RECORD_HEADER_VALUES)) {
+                for (int i = 0; i < headers.length; i++) {
+                    headers[i] = encryptRecordHeaderValue(keyContext, headers[i]);
+                }
+            }
+            Header[] transformedHeaders = prependToHeaders(headers, fieldsHeader, dekHeader);
+            receiver.accept(kafkaRecord, transformedValue, transformedHeaders);
+
+            if (valueCiphertext != null) {
+                valueCiphertext.rewind();
+            }
+        });
+    }
+
+    private Header encryptRecordHeaderValue(KeyContext keyContext, Header header) {
+        byte[] headerValue = header.value();
+        ByteBuffer plaintext = ByteBuffer.wrap(headerValue);
+        var ciphertext = ByteBuffer.allocate(keyContext.encodedSize(headerValue.length));
+        keyContext.encode(plaintext, ciphertext);
+        ciphertext.flip();
+        return new RecordHeader(header.key(), ciphertext.array());
+    }
+
+    @Nullable
+    private ByteBuffer encryptRecordValue(KeyContext keyContext,
+                                          Record kafkaRecord,
+                                          ByteBuffer valueCiphertext) {
+        ByteBuffer transformedValue;
+        if (!kafkaRecord.hasValue()) {
+            transformedValue = null;
+        }
+        else {
+            ByteBuffer plaintext = kafkaRecord.value();
+            keyContext.encodedSize(kafkaRecord.valueSize());
+            keyContext.encode(plaintext, valueCiphertext);
+            valueCiphertext.flip();
+            transformedValue = valueCiphertext;
+        }
+        return transformedValue;
+    }
+
+    static @NonNull RecordHeader createEncryptedFieldsHeader(@NonNull Set<RecordField> recordFields) {
+        return new RecordHeader(FIELDS_HEADER_NAME, new byte[]{ RecordField.toBits(recordFields) });
+    }
+
+    static @NonNull RecordHeader createEdekHeader(@NonNull KeyContext keyContext) {
+        return new RecordHeader(DEK_HEADER_NAME, keyContext.prefix());
+    }
+
+    @NonNull
+    static Header[] prependToHeaders(Header[] oldHeaders, @NonNull RecordHeader... additionalHeaders) {
+        if (additionalHeaders.length == 0) {
+            return oldHeaders;
+        }
+        if (oldHeaders == null || oldHeaders.length == 0) {
+            return additionalHeaders;
+        }
+        Header[] newHeaders = new Header[oldHeaders.length + additionalHeaders.length];
+        System.arraycopy(additionalHeaders, 0, newHeaders, 0, additionalHeaders.length);
+        System.arraycopy(oldHeaders, 0, newHeaders, additionalHeaders.length, oldHeaders.length);
+        return newHeaders;
+    }
+
+    @NonNull
+    static Header[] removeInitialHeaders(@NonNull Header[] oldHeaders, int numToRemove) {
+        if (numToRemove < 0) {
+            throw new IllegalArgumentException();
+        }
+        Header[] newHeaders = new Header[oldHeaders.length - numToRemove];
+        if (newHeaders.length > 0) {
+            System.arraycopy(oldHeaders, numToRemove, newHeaders, 0, newHeaders.length);
+        }
+        return newHeaders;
+    }
+
+    static Set<RecordField> encryptedFields(Record kafkaRecord) {
+        for (Header header : kafkaRecord.headers()) {
+            if (FIELDS_HEADER_NAME.equals(header.key())) {
+                return RecordField.fromBits(header.value()[0]);
+            }
+        }
+        return EnumSet.noneOf(RecordField.class);
+    }
+
+    static RecordHeader dek(Record kafkaRecord) {
+        for (Header header : kafkaRecord.headers()) {
+            if (DEK_HEADER_NAME.equals(header.key())) {
+                return (RecordHeader) header;
+            }
+        }
+        throw new IllegalStateException();
+    }
+
+    private CompletionStage<AesGcmEncryptor> getOrCacheDecryptor(RecordHeader dekHeader,
+                                                                 K kekId,
+                                                                 E edek) {
+        return decryptorCache.compute(dekHeader, (k, v) -> {
+            if (v == null) {
+                return kms.decryptEdek(kekId, edek)
+                        .thenApply(AesGcmEncryptor::forDecrypt).toCompletableFuture();
+                // TODO what happens if the CS complete exceptionally
+                // TODO what happens if the CS doesn't complete at all in a reasonably time frame?
+            }
+            else {
+                return v;
+            }
+        });
+    }
+
+    @NonNull
+    @Override
+    public CompletionStage<Void> decrypt(@NonNull List<? extends Record> records,
+                                         @NonNull Receiver receiver) {
+        List<CompletionStage<Void>> futures = new ArrayList<>(records.size());
+        for (var kafkaRecord : records) {
+            var encryptedFields = encryptedFields(kafkaRecord);
+            if (encryptedFields.isEmpty()) {
+                receiver.accept(kafkaRecord, kafkaRecord.value(), kafkaRecord.headers());
+                futures.add(CompletableFuture.completedFuture(null));
+            }
+            else {
+                // right now (because we only support topic name based kek selection) noce we've resolve the first value we
+                // can keep the lock and process all the records
+                var x = resolveEncryptor(kafkaRecord).thenAccept(encryptor -> {
+                    decryptRecord(receiver, kafkaRecord, encryptor, encryptedFields);
+
+                });
+                futures.add(x);
+            }
+        }
+
+        return io.kroxylicious.filter.encryption.EnvelopeEncryptionFilter.join(futures).thenAccept(list -> {
+        });
+    }
+
+    @SuppressWarnings("java:S2445")
+    private void decryptRecord(@NonNull Receiver receiver, Record kafkaRecord, AesGcmEncryptor encryptor, Set<RecordField> encryptedFields) {
+        ByteBuffer decryptedValue;
+        var headers = removeInitialHeaders(kafkaRecord.headers(), 2);
+        synchronized (encryptor) {
+            if (encryptedFields.contains(RecordField.RECORD_VALUE)) {
+                decryptedValue = decryptRecordValue(kafkaRecord, encryptor);
+            }
+            else {
+                decryptedValue = kafkaRecord.value();
+            }
+            if (encryptedFields.contains(RecordField.RECORD_HEADER_VALUES)) {
+                for (int i = 0; i < headers.length; i++) {
+                    headers[i] = decryptRecordHeader(headers[i], encryptor);
+                }
+            }
+        }
+        receiver.accept(kafkaRecord, decryptedValue, headers);
+    }
+
+    private CompletionStage<AesGcmEncryptor> resolveEncryptor(Record kafkaRecord) {
+        var dekHeader = dek(kafkaRecord);
+        var buffer = ByteBuffer.wrap(dekHeader.value());
+        var kekLength = buffer.getShort();
+        int origLimit = buffer.limit();
+        buffer.limit(buffer.position() + kekLength);
+        var kekId = kekIdSerde.deserialize(buffer);
+        buffer.limit(origLimit);
+        var edekLength = buffer.getShort();
+        buffer.limit(buffer.position() + edekLength);
+        var edek = edekSerde.deserialize(buffer);
+        buffer.rewind();
+        return getOrCacheDecryptor(dekHeader, kekId, edek);
+    }
+
+    private Header decryptRecordHeader(Header header, AesGcmEncryptor encryptor) {
+        var ciphertext = ByteBuffer.wrap(header.value());
+        var plaintext = ciphertext.duplicate();
+        encryptor.decrypt(ciphertext, plaintext);
+        plaintext.flip();
+        byte[] value = new byte[plaintext.limit()];
+        plaintext.get(value);
+        return new RecordHeader(header.key(), value);
+    }
+
+    private ByteBuffer decryptRecordValue(Record kafkaRecord, AesGcmEncryptor encryptor) {
+        if (!kafkaRecord.hasValue()) {
+            return kafkaRecord.value();
+        }
+        else {
+            ByteBuffer ciphertext = kafkaRecord.value();
+            ByteBuffer plaintext = ciphertext.duplicate();
+            encryptor.decrypt(ciphertext, plaintext);
+            plaintext.flip();
+            return plaintext;
+        }
+    }
+
+}
diff --git a/kroxylicious-filters/kroxylicious-encryption/src/main/java/io/kroxylicious/filter/encryption/inband/KeyContext.java b/kroxylicious-filters/kroxylicious-encryption/src/main/java/io/kroxylicious/filter/encryption/inband/KeyContext.java
new file mode 100644
index 00000000..5e5e59b7
--- /dev/null
+++ b/kroxylicious-filters/kroxylicious-encryption/src/main/java/io/kroxylicious/filter/encryption/inband/KeyContext.java
@@ -0,0 +1,80 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.filter.encryption.inband;
+
+import java.nio.ByteBuffer;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import javax.security.auth.DestroyFailedException;
+import javax.security.auth.Destroyable;
+
+import edu.umd.cs.findbugs.annotations.NonNull;
+
+/**
+ * A DekContext encapsulates an encryptor.
+ */
+final class KeyContext implements Destroyable {
+    private final AesGcmEncryptor encryptor;
+    private final byte[] prefix;
+    private final long encryptionExpiryNanos;
+    private final AtomicInteger remainingEncryptions;
+
+    KeyContext(@NonNull ByteBuffer prefix,
+               long encryptionExpiryNanos,
+               int maxEncryptions,
+               @NonNull AesGcmEncryptor encryptor) {
+        if (maxEncryptions <= 0) {
+            throw new IllegalArgumentException();
+        }
+        this.prefix = prefix.array();
+        this.encryptionExpiryNanos = encryptionExpiryNanos;
+        this.remainingEncryptions = new AtomicInteger(maxEncryptions);
+        this.encryptor = encryptor;
+    }
+
+    public byte[] prefix() {
+        return prefix;
+    }
+
+    public boolean isExpiredForEncryption(long nanoTime) {
+        return nanoTime > encryptionExpiryNanos;
+    }
+
+    public boolean hasAtLeastRemainingEncryptions(int numEncryptions) {
+        if (numEncryptions <= 0) {
+            throw new IllegalArgumentException();
+        }
+        return remainingEncryptions.getAndAdd(-numEncryptions) >= numEncryptions;
+    }
+
+    /**
+     * Returns the size of the encoding of a plaintext of the given size
+     * @param plaintextSize The plaintext.
+     * @return The size, in bytes, of a plaintext.
+     */
+    public int encodedSize(int plaintextSize) {
+        return encryptor.outputSize(plaintextSize);
+    }
+
+    /**
+     * Encode the key metadata and the ciphertext of the given {@code plaintext} to the given {@code output},
+     * which should have at least {@link #encodedSize(int) encodedSize(plaintext)} bytes {@linkplain ByteBuffer#remaining() remaining}.
+     * @param plaintext The plaintext
+     * @param output The output buffer
+     */
+    public void encode(@NonNull ByteBuffer plaintext, @NonNull ByteBuffer output) {
+        if (remainingEncryptions.get() < 0) {
+            throw new ExhaustedDekException("No more encryptions");
+        }
+        encryptor.encrypt(plaintext, output);
+    }
+
+    @Override
+    public void destroy() throws DestroyFailedException {
+        encryptor.destroy();
+    }
+}
diff --git a/kroxylicious-filters/kroxylicious-encryption/src/main/resources/META-INF/services/io.kroxylicious.filter.encryption.KekSelectorService b/kroxylicious-filters/kroxylicious-encryption/src/main/resources/META-INF/services/io.kroxylicious.filter.encryption.KekSelectorService
new file mode 100644
index 00000000..4b077633
--- /dev/null
+++ b/kroxylicious-filters/kroxylicious-encryption/src/main/resources/META-INF/services/io.kroxylicious.filter.encryption.KekSelectorService
@@ -0,0 +1,1 @@
+io.kroxylicious.filter.encryption.TemplateKekSelector
\ No newline at end of file
diff --git a/kroxylicious-filters/kroxylicious-encryption/src/main/resources/META-INF/services/io.kroxylicious.proxy.filter.FilterFactory b/kroxylicious-filters/kroxylicious-encryption/src/main/resources/META-INF/services/io.kroxylicious.proxy.filter.FilterFactory
new file mode 100644
index 00000000..8c30b394
--- /dev/null
+++ b/kroxylicious-filters/kroxylicious-encryption/src/main/resources/META-INF/services/io.kroxylicious.proxy.filter.FilterFactory
@@ -0,0 +1,1 @@
+io.kroxylicious.filter.encryption.EnvelopeEncryption
\ No newline at end of file
diff --git a/kroxylicious-filters/pom.xml b/kroxylicious-filters/pom.xml
index 6f3121e9..4d39b91c 100644
--- a/kroxylicious-filters/pom.xml
+++ b/kroxylicious-filters/pom.xml
@@ -23,6 +23,7 @@
     <packaging>pom</packaging>
 
     <modules>
+        <module>kroxylicious-encryption</module>
         <module>kroxylicious-multitenant</module>
         <module>kroxylicious-record-validation</module>
         <module>kroxylicious-simple-transform</module>
@@ -48,6 +49,10 @@
                                     <includes>
                                         <include>io.kroxylicious:kroxylicious-api</include>
                                         <include>io.kroxylicious:kroxylicious-filter-test-support</include>
+                                        <include>io.kroxylicious:kroxylicious-kms</include>
+                                        <include>io.kroxylicious:kroxylicious-kms-provider-kroxylicious-inmemory</include>
+                                        <include>io.kroxylicious:kroxylicious-unit-test-support</include>
+                                        <include>io.kroxylicious:kroxylicious-integration-test-support</include>
                                     </includes>
                                 </bannedDependencies>
                             </rules>


Test patch:
diff --git a/kroxylicious-filters/kroxylicious-encryption/src/test/java/io/kroxylicious/filter/encryption/EncryptionSchemeTest.java b/kroxylicious-filters/kroxylicious-encryption/src/test/java/io/kroxylicious/filter/encryption/EncryptionSchemeTest.java
new file mode 100644
index 00000000..2574aeb6
--- /dev/null
+++ b/kroxylicious-filters/kroxylicious-encryption/src/test/java/io/kroxylicious/filter/encryption/EncryptionSchemeTest.java
@@ -0,0 +1,37 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.filter.encryption;
+
+import java.util.EnumSet;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+class EncryptionSchemeTest {
+
+    @Test
+    void shouldRejectInvalidConstructorArgs() {
+        EnumSet<RecordField> nonEmpty = EnumSet.of(RecordField.RECORD_VALUE);
+        var empty = EnumSet.noneOf(RecordField.class);
+        assertThrows(NullPointerException.class, () -> new EncryptionScheme<>(null, nonEmpty));
+        Object kekId = new Object();
+        assertThrows(NullPointerException.class, () -> new EncryptionScheme<>(kekId, null));
+        assertThrows(IllegalArgumentException.class, () -> new EncryptionScheme<>(kekId, empty));
+    }
+
+    @Test
+    void shouldAcceptValidConstructorArgs() {
+        EnumSet<RecordField> nonEmpty = EnumSet.of(RecordField.RECORD_VALUE);
+        Object kekId = new Object();
+        var es = new EncryptionScheme<>(kekId, nonEmpty);
+        assertEquals(kekId, es.kekId());
+        assertEquals(nonEmpty, es.recordFields());
+    }
+
+}
diff --git a/kroxylicious-filters/kroxylicious-encryption/src/test/java/io/kroxylicious/filter/encryption/EnvelopeEncryptionFilterTest.java b/kroxylicious-filters/kroxylicious-encryption/src/test/java/io/kroxylicious/filter/encryption/EnvelopeEncryptionFilterTest.java
new file mode 100644
index 00000000..c64c5aca
--- /dev/null
+++ b/kroxylicious-filters/kroxylicious-encryption/src/test/java/io/kroxylicious/filter/encryption/EnvelopeEncryptionFilterTest.java
@@ -0,0 +1,304 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.filter.encryption;
+
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Consumer;
+
+import org.apache.kafka.common.header.Header;
+import org.apache.kafka.common.header.internals.RecordHeader;
+import org.apache.kafka.common.message.FetchResponseData;
+import org.apache.kafka.common.message.ProduceRequestData;
+import org.apache.kafka.common.message.ProduceResponseData;
+import org.apache.kafka.common.message.RequestHeaderData;
+import org.apache.kafka.common.message.ResponseHeaderData;
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.record.CompressionType;
+import org.apache.kafka.common.record.MemoryRecordsBuilder;
+import org.apache.kafka.common.record.Record;
+import org.apache.kafka.common.record.RecordBatch;
+import org.apache.kafka.common.record.SimpleRecord;
+import org.apache.kafka.common.record.TimestampType;
+import org.apache.kafka.common.utils.ByteBufferOutputStream;
+import org.assertj.core.api.AbstractAssert;
+import org.assertj.core.api.InstanceOfAssertFactories;
+import org.assertj.core.matcher.AssertionMatcher;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Captor;
+import org.mockito.Mock;
+import org.mockito.hamcrest.MockitoHamcrest;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import io.kroxylicious.filter.encryption.inband.TestingRecord;
+import io.kroxylicious.proxy.filter.FilterContext;
+import io.kroxylicious.proxy.filter.RequestFilterResult;
+
+import edu.umd.cs.findbugs.annotations.NonNull;
+
+import static io.kroxylicious.filter.encryption.ProduceRequestDataCondition.hasRecordsForTopic;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.ArgumentMatchers.anySet;
+import static org.mockito.ArgumentMatchers.assertArg;
+import static org.mockito.Mock.Strictness.LENIENT;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+class EnvelopeEncryptionFilterTest {
+
+    public static final String UNENCRYPTED_TOPIC = "unencrypted";
+    public static final String ENCRYPTED_TOPIC = "encrypt_me";
+    public static final String KEK_ID_1 = "KEK_ID_1";
+    public static final String ENCRYPTED_MESSAGE = "xslkajfd;ljsaefjjKLDJlkDSJFLJK';,kSDKF'";
+
+    @Mock(strictness = LENIENT)
+    KeyManager<String> keyManager;
+
+    @Mock(strictness = LENIENT)
+    TopicNameBasedKekSelector<String> kekSelector;
+
+    @Mock(strictness = LENIENT)
+    private FilterContext context;
+
+    @Captor
+    private ArgumentCaptor<ApiMessage> apiMessageCaptor;
+
+    private EnvelopeEncryptionFilter<String> encryptionFilter;
+
+    @BeforeEach
+    void setUp() {
+        when(context.forwardRequest(any(RequestHeaderData.class), apiMessageCaptor.capture())).then(invocationOnMock -> {
+            final RequestFilterResult filterResult = mock(RequestFilterResult.class);
+            return CompletableFuture.completedFuture(filterResult);
+        });
+
+        when(context.createByteBufferOutputStream(anyInt())).thenAnswer(invocationOnMock -> {
+            final int capacity = invocationOnMock.getArgument(0);
+            return new ByteBufferOutputStream(capacity);
+        });
+
+        final Map<String, String> topicNameToKekId = new HashMap<>();
+        topicNameToKekId.put(UNENCRYPTED_TOPIC, null);
+        topicNameToKekId.put(ENCRYPTED_TOPIC, KEK_ID_1);
+        when(kekSelector.selectKek(anySet())).thenReturn(CompletableFuture.completedFuture(topicNameToKekId));
+
+        when(keyManager.encrypt(any(), anyList(), any(Receiver.class))).thenAnswer(invocationOnMock -> {
+            final List<? extends Record> actualRecords = invocationOnMock.getArgument(1);
+            final Receiver receiver = invocationOnMock.getArgument(2);
+            for (Record actualRecord : actualRecords) {
+                receiver.accept(actualRecord, ByteBuffer.allocate(actualRecord.sizeInBytes()), new Header[0]);
+            }
+            return CompletableFuture.completedFuture(null);
+        });
+
+        when(keyManager.decrypt(anyList(), any(Receiver.class))).thenAnswer(invocationOnMock -> {
+            final List<? extends Record> actualRecords = invocationOnMock.getArgument(0);
+            final Receiver receiver = invocationOnMock.getArgument(1);
+            for (Record actualRecord : actualRecords) {
+                receiver.accept(actualRecord, ByteBuffer.allocate(actualRecord.sizeInBytes()), new Header[0]);
+            }
+            return CompletableFuture.completedFuture(null);
+        });
+
+        encryptionFilter = new EnvelopeEncryptionFilter<>(keyManager, kekSelector);
+    }
+
+    @Test
+    void shouldNotEncryptTopicWithoutKeyId() {
+        // Given
+        final ProduceRequestData produceRequestData = buildProduceRequestData(new Payload(UNENCRYPTED_TOPIC, "Hello World!"));
+
+        // When
+        encryptionFilter.onProduceRequest(ProduceRequestData.HIGHEST_SUPPORTED_VERSION, new RequestHeaderData(), produceRequestData, context);
+
+        // Then
+        verify(keyManager, never()).encrypt(any(), anyList(), any(Receiver.class));
+    }
+
+    @Test
+    void shouldEncryptTopicWithKeyId() {
+        // Given
+        final ProduceRequestData produceRequestData = buildProduceRequestData(new Payload(ENCRYPTED_TOPIC, "Hello World!"));
+
+        // When
+        encryptionFilter.onProduceRequest(ProduceRequestData.HIGHEST_SUPPORTED_VERSION, new RequestHeaderData(), produceRequestData, context);
+
+        // Then
+        verify(keyManager).encrypt(any(), anyList(), any(Receiver.class));
+    }
+
+    @Test
+    void shouldOnlyEncryptTopicWithKeyId() {
+        // Given
+        final Payload encryptedPayload = new Payload(ENCRYPTED_TOPIC, "Hello Ciphertext World!");
+        final ProduceRequestData produceRequestData = buildProduceRequestData(encryptedPayload,
+                new Payload(UNENCRYPTED_TOPIC, "Hello Plaintext World!"));
+
+        // When
+        encryptionFilter.onProduceRequest(ProduceRequestData.HIGHEST_SUPPORTED_VERSION, new RequestHeaderData(), produceRequestData, context);
+
+        // Then
+        verify(keyManager).encrypt(any(),
+                argThat(records -> assertThat(records)
+                        .hasSize(1)
+                        .allSatisfy(record -> assertThat(record.value()).isEqualByComparingTo(encryptedPayload.messageBytes()))),
+                any());
+    }
+
+    @Test
+    void shouldPassThroughUnEncryptedRecords() {
+        // Given
+        final FetchResponseData fetchResponseData = buildFetchResponseData(new Payload(UNENCRYPTED_TOPIC, "Hello Plaintext World!"));
+
+        // When
+        encryptionFilter.onFetchResponse(ProduceResponseData.HIGHEST_SUPPORTED_VERSION,
+                new ResponseHeaderData(), fetchResponseData, context);
+
+        // Then
+        verify(context).forwardResponse(any(ResponseHeaderData.class), assertArg(actualFetchResponse -> assertThat(actualFetchResponse)
+                .isInstanceOf(FetchResponseData.class).isEqualTo(fetchResponseData)));
+    }
+
+    @Test
+    void shouldDecryptEncryptedRecords() {
+        // Given
+        final FetchResponseData encryptedFetchResponse = buildFetchResponseData(new Payload(ENCRYPTED_TOPIC, ENCRYPTED_MESSAGE));
+        final Payload plainTextPayload = new Payload(ENCRYPTED_TOPIC, "Hello Plaintext World!");
+
+        when(keyManager.decrypt(assertArg(records -> assertThat(records).hasSize(1)), any(Receiver.class))).thenAnswer(invocationOnMock -> {
+            final Receiver receiver = invocationOnMock.getArgument(1);
+            receiver.accept(new TestingRecord(plainTextPayload.messageBytes()), plainTextPayload.messageBytes(), Record.EMPTY_HEADERS);
+
+            return CompletableFuture.completedFuture(null);
+        });
+
+        // When
+        encryptionFilter.onFetchResponse(ProduceResponseData.HIGHEST_SUPPORTED_VERSION,
+                new ResponseHeaderData(), encryptedFetchResponse, context);
+
+        // Then
+        verify(context).forwardResponse(any(ResponseHeaderData.class), assertArg(actualFetchResponse -> assertThat(actualFetchResponse)
+                .isInstanceOf(FetchResponseData.class)
+                .asInstanceOf(InstanceOfAssertFactories.type(FetchResponseData.class))
+        // .has(new FetchResponseDataCondition(fetchResponseData -> true)) //This is where the new conditions from https://github.com/kroxylicious/kroxylicious/pull/756 come in
+        ));
+    }
+
+    @Test
+    void shouldEncryptTopic() {
+        // Given
+        final Payload encryptedPayload = new Payload(ENCRYPTED_TOPIC, "Hello Ciphertext World!");
+        final ProduceRequestData produceRequestData = buildProduceRequestData(encryptedPayload,
+                new Payload(UNENCRYPTED_TOPIC, "Hello Plaintext World!"));
+
+        // When
+        encryptionFilter.onProduceRequest(ProduceRequestData.HIGHEST_SUPPORTED_VERSION, new RequestHeaderData(), produceRequestData, context);
+
+        // Then
+        verify(context).forwardRequest(any(), argThat(request -> assertThat(request).isInstanceOf(ProduceRequestData.class)
+                .asInstanceOf(InstanceOfAssertFactories.type(ProduceRequestData.class))
+                .is(hasRecordsForTopic(ENCRYPTED_TOPIC))));
+    }
+
+    private static FetchResponseData buildFetchResponseData(Payload... payloads) {
+        final FetchResponseData fetchResponseData = new FetchResponseData();
+        var stream = new ByteBufferOutputStream(ByteBuffer.allocate(1000));
+        for (Payload payload : payloads) {
+            var recordsBuilder = memoryRecordsBuilderForStream(stream);
+            final FetchResponseData.FetchableTopicResponse fetchableTopicResponse = new FetchResponseData.FetchableTopicResponse();
+            fetchableTopicResponse.setTopic(payload.topicName());
+            final FetchResponseData.PartitionData partitionData = new FetchResponseData.PartitionData();
+            recordsBuilder.append(new SimpleRecord(System.currentTimeMillis(), null, payload.message.getBytes(StandardCharsets.UTF_8)));
+            partitionData.setRecords(recordsBuilder.build());
+            fetchableTopicResponse.partitions().add(partitionData);
+
+            fetchResponseData.responses().add(fetchableTopicResponse);
+        }
+        return fetchResponseData;
+    }
+
+    private static ProduceRequestData buildProduceRequestData(Payload... payloads) {
+        var requestData = new ProduceRequestData();
+
+        // Build records from stream
+        var stream = new ByteBufferOutputStream(ByteBuffer.allocate(1000));
+        var topics = new ProduceRequestData.TopicProduceDataCollection();
+
+        for (Payload payload : payloads) {
+            var recordsBuilder = memoryRecordsBuilderForStream(stream);
+
+            // Create record Headers
+            Header header = new RecordHeader("myKey", "myValue".getBytes());
+
+            // Add transformValue as buffer to records
+            recordsBuilder.append(RecordBatch.NO_TIMESTAMP, null, payload.messageBytes(), new Header[]{ header });
+            var records = recordsBuilder.build();
+            // Build partitions from built records
+            var partitions = new ArrayList<ProduceRequestData.PartitionProduceData>();
+            var partitionData = new ProduceRequestData.PartitionProduceData();
+            partitionData.setRecords(records);
+            partitions.add(partitionData);
+            // Build topics from built partitions
+
+            var topicData = new ProduceRequestData.TopicProduceData();
+            topicData.setPartitionData(partitions);
+            topicData.setName(payload.topicName);
+            topics.add(topicData);
+        }
+
+        // Add built topics to ProduceRequestData object so that we can return it
+        requestData.setTopicData(topics);
+        return requestData;
+    }
+
+    @NonNull
+    private static MemoryRecordsBuilder memoryRecordsBuilderForStream(ByteBufferOutputStream stream) {
+        return new MemoryRecordsBuilder(stream, RecordBatch.CURRENT_MAGIC_VALUE, CompressionType.NONE, TimestampType.CREATE_TIME, 0,
+                RecordBatch.NO_TIMESTAMP, RecordBatch.NO_PRODUCER_ID, RecordBatch.NO_PRODUCER_EPOCH, RecordBatch.NO_SEQUENCE, false, false,
+                RecordBatch.NO_PARTITION_LEADER_EPOCH, stream.remaining());
+    }
+
+    private record Payload(String topicName, String message) {
+        ByteBuffer messageBytes() {
+            return ByteBuffer.wrap(message.getBytes(StandardCharsets.UTF_8)).position(0);
+        }
+    }
+
+    public static <T> T argThat(Consumer<T> assertions) {
+        return MockitoHamcrest.argThat(new AssertionMatcher<>() {
+
+            String underlyingDescription;
+
+            @Override
+            public void assertion(T actual) throws AssertionError {
+                AbstractAssert.setDescriptionConsumer(description -> underlyingDescription = description.value());
+                assertions.accept(actual);
+            }
+
+            @Override
+            public void describeTo(org.hamcrest.Description description) {
+                super.describeTo(description);
+                description.appendValue(Objects.requireNonNullElse(underlyingDescription, "custom argument matcher"));
+            }
+        });
+    }
+}
diff --git a/kroxylicious-filters/kroxylicious-encryption/src/test/java/io/kroxylicious/filter/encryption/EnvelopeEncryptionTest.java b/kroxylicious-filters/kroxylicious-encryption/src/test/java/io/kroxylicious/filter/encryption/EnvelopeEncryptionTest.java
new file mode 100644
index 00000000..a49d000f
--- /dev/null
+++ b/kroxylicious-filters/kroxylicious-encryption/src/test/java/io/kroxylicious/filter/encryption/EnvelopeEncryptionTest.java
@@ -0,0 +1,43 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.filter.encryption;
+
+import org.junit.jupiter.api.Test;
+
+import io.kroxylicious.kms.service.Kms;
+import io.kroxylicious.kms.service.KmsService;
+import io.kroxylicious.proxy.filter.FilterFactoryContext;
+
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+
+class EnvelopeEncryptionTest {
+
+    @Test
+    void shouldInitAndCreateFilter() {
+        EnvelopeEncryption.Config config = new EnvelopeEncryption.Config("KMS", null, "SELECTOR", null);
+        var ee = new EnvelopeEncryption<>();
+        var fc = mock(FilterFactoryContext.class);
+        var kmsService = mock(KmsService.class);
+        var kms = mock(Kms.class);
+        var kekSelectorService = mock(KekSelectorService.class);
+        var kekSelector = mock(TopicNameBasedKekSelector.class);
+
+        doReturn(kmsService).when(fc).pluginInstance(KmsService.class, "KMS");
+        doReturn(kms).when(kmsService).buildKms(any());
+
+        doReturn(kekSelectorService).when(fc).pluginInstance(KekSelectorService.class, "SELECTOR");
+        doReturn(kekSelector).when(kekSelectorService).buildSelector(any(), any());
+
+        ee.initialize(fc, config);
+        var filter = ee.createFilter(fc, config);
+        assertNotNull(filter);
+    }
+
+}
diff --git a/kroxylicious-filters/kroxylicious-encryption/src/test/java/io/kroxylicious/filter/encryption/ProduceRequestDataCondition.java b/kroxylicious-filters/kroxylicious-encryption/src/test/java/io/kroxylicious/filter/encryption/ProduceRequestDataCondition.java
new file mode 100644
index 00000000..ad659d28
--- /dev/null
+++ b/kroxylicious-filters/kroxylicious-encryption/src/test/java/io/kroxylicious/filter/encryption/ProduceRequestDataCondition.java
@@ -0,0 +1,28 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.filter.encryption;
+
+import org.apache.kafka.common.message.ProduceRequestData;
+import org.assertj.core.api.Condition;
+
+public class ProduceRequestDataCondition extends Condition<ProduceRequestData> {
+
+    private final String topicName;
+
+    public ProduceRequestDataCondition(String topicName) {
+        this.topicName = topicName;
+    }
+
+    public static Condition<ProduceRequestData> hasRecordsForTopic(String topicName) {
+        return new ProduceRequestDataCondition(topicName);
+    }
+
+    @Override
+    public boolean matches(ProduceRequestData value) {
+        return value.topicData().stream().anyMatch(topicProduceData -> topicProduceData.name().equals(topicName));
+    }
+}
\ No newline at end of file
diff --git a/kroxylicious-filters/kroxylicious-encryption/src/test/java/io/kroxylicious/filter/encryption/RecordFieldTest.java b/kroxylicious-filters/kroxylicious-encryption/src/test/java/io/kroxylicious/filter/encryption/RecordFieldTest.java
new file mode 100644
index 00000000..ede910e9
--- /dev/null
+++ b/kroxylicious-filters/kroxylicious-encryption/src/test/java/io/kroxylicious/filter/encryption/RecordFieldTest.java
@@ -0,0 +1,31 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.filter.encryption;
+
+import java.util.EnumSet;
+import java.util.List;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+class RecordFieldTest {
+
+    @Test
+    void serializationShouldRoundTrip() {
+        for (var recordValue : List.of(
+                EnumSet.of(RecordField.RECORD_VALUE),
+                EnumSet.of(RecordField.RECORD_HEADER_VALUES),
+                EnumSet.of(RecordField.RECORD_HEADER_VALUES, RecordField.RECORD_VALUE),
+                EnumSet.allOf(RecordField.class),
+                EnumSet.noneOf(RecordField.class))) {
+            var b = RecordField.toBits(recordValue);
+            assertEquals(recordValue, RecordField.fromBits(b), recordValue.toString());
+        }
+    }
+
+}
\ No newline at end of file
diff --git a/kroxylicious-filters/kroxylicious-encryption/src/test/java/io/kroxylicious/filter/encryption/TemplateKekSelectorTest.java b/kroxylicious-filters/kroxylicious-encryption/src/test/java/io/kroxylicious/filter/encryption/TemplateKekSelectorTest.java
new file mode 100644
index 00000000..75e1aaac
--- /dev/null
+++ b/kroxylicious-filters/kroxylicious-encryption/src/test/java/io/kroxylicious/filter/encryption/TemplateKekSelectorTest.java
@@ -0,0 +1,58 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.filter.encryption;
+
+import java.util.Set;
+import java.util.concurrent.ExecutionException;
+
+import org.junit.jupiter.api.Test;
+
+import io.kroxylicious.kms.provider.kroxylicious.inmemory.InMemoryKms;
+import io.kroxylicious.kms.provider.kroxylicious.inmemory.UnitTestingKmsService;
+import io.kroxylicious.kms.service.Kms;
+
+import edu.umd.cs.findbugs.annotations.NonNull;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+class TemplateKekSelectorTest {
+
+    @Test
+    void shouldRejectUnknownPlaceholders() {
+        var e = assertThrows(IllegalArgumentException.class, () -> getSelector(null, "foo-${topicId}-bar"));
+        assertEquals("Unknown template parameter: topicId", e.getMessage());
+    }
+
+    @Test
+    void shouldResolveWhenAliasExists() throws ExecutionException, InterruptedException {
+        InMemoryKms kms = UnitTestingKmsService.newInstance().buildKms(new UnitTestingKmsService.Config());
+        var selector = getSelector(kms, "topic-${topicName}");
+
+        var kek = kms.generateKey();
+        kms.createAlias(kek, "topic-my-topic");
+        var map = selector.selectKek(Set.of("my-topic")).toCompletableFuture().get();
+        assertEquals(kek, map.get("my-topic"));
+    }
+
+    @Test
+    void shouldThrowWhenAliasDoesNotExist() throws ExecutionException, InterruptedException {
+        InMemoryKms kms = UnitTestingKmsService.newInstance().buildKms(new UnitTestingKmsService.Config());
+        var selector = getSelector(kms, "topic-${topicName}");
+
+        var map = selector.selectKek(Set.of("my-topic")).toCompletableFuture().get();
+        assertNull(map.get("my-topic"));
+    }
+
+    @NonNull
+    private <K> TopicNameBasedKekSelector<K> getSelector(Kms<K, ?> kms, String template) {
+        var config = new TemplateKekSelector.Config(template);
+        return new TemplateKekSelector<K>().buildSelector(kms, config);
+    }
+
+}
diff --git a/kroxylicious-filters/kroxylicious-encryption/src/test/java/io/kroxylicious/filter/encryption/inband/AesGcmEncryptorTest.java b/kroxylicious-filters/kroxylicious-encryption/src/test/java/io/kroxylicious/filter/encryption/inband/AesGcmEncryptorTest.java
new file mode 100644
index 00000000..ea24fdab
--- /dev/null
+++ b/kroxylicious-filters/kroxylicious-encryption/src/test/java/io/kroxylicious/filter/encryption/inband/AesGcmEncryptorTest.java
@@ -0,0 +1,184 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.filter.encryption.inband;
+
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.security.InvalidKeyException;
+import java.security.NoSuchAlgorithmException;
+import java.security.SecureRandom;
+
+import javax.crypto.AEADBadTagException;
+import javax.crypto.KeyGenerator;
+import javax.crypto.SecretKey;
+import javax.crypto.ShortBufferException;
+import javax.crypto.spec.SecretKeySpec;
+import javax.security.auth.DestroyFailedException;
+
+import org.junit.jupiter.api.Test;
+
+import io.kroxylicious.filter.encryption.EncryptionException;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class AesGcmEncryptorTest {
+
+    @Test
+    void shouldRejectInvalidUsagePattern() throws NoSuchAlgorithmException {
+        var keygen = KeyGenerator.getInstance("AES");
+        SecretKey key = keygen.generateKey();
+        var enc = AesGcmEncryptor.forEncrypt(new AesGcmIvGenerator(new SecureRandom()), key);
+        var plaintext = "hello, world!";
+        byte[] plaintextBytes = plaintext.getBytes(StandardCharsets.UTF_8);
+
+        // A call to encrypt without prior call to outputSize
+        assertThrows(IllegalStateException.class, () -> enc.encrypt(null, null));
+
+        var size = enc.outputSize(plaintextBytes.length);
+        var ciphertext = ByteBuffer.allocate(size);
+        enc.encrypt(ByteBuffer.wrap(plaintextBytes), ciphertext);
+        // Two calls to encrypt in a row
+        assertThrows(IllegalStateException.class, () -> enc.encrypt(null, null));
+
+        assertThrows(IllegalStateException.class, () -> enc.decrypt(null, null));
+    }
+
+    @Test
+    void shouldDestroy() throws NoSuchAlgorithmException, DestroyFailedException {
+        var keygen = KeyGenerator.getInstance("AES");
+        SecretKey key = keygen.generateKey();
+        var enc = AesGcmEncryptor.forEncrypt(new AesGcmIvGenerator(new SecureRandom()), key);
+        assertThrows(DestroyFailedException.class, enc::destroy);
+        // ^^ this is not the behaviour we want, but it's the behaviour we get (from the in memory KMS at least)
+        assertTrue(enc.isDestroyed());
+
+        enc.destroy();
+        // once destroyed we expect a 2nd call to not throw
+    }
+
+    @Test
+    void shouldRoundTrip() throws NoSuchAlgorithmException {
+        var keygen = KeyGenerator.getInstance("AES");
+        SecretKey key = keygen.generateKey();
+        var enc = AesGcmEncryptor.forEncrypt(new AesGcmIvGenerator(new SecureRandom()), key);
+        var plaintext = "hello, world!";
+        byte[] plaintextBytes = plaintext.getBytes(StandardCharsets.UTF_8);
+        var size = enc.outputSize(plaintextBytes.length);
+        var ciphertext = ByteBuffer.allocate(size);
+        enc.encrypt(ByteBuffer.wrap(plaintextBytes), ciphertext);
+        assertFalse(ciphertext.hasRemaining());
+        ciphertext.flip();
+
+        var roundTripped = ByteBuffer.allocate(plaintextBytes.length);
+        var dec = AesGcmEncryptor.forDecrypt(key);
+        dec.decrypt(ciphertext, roundTripped);
+        assertEquals(plaintext, new String(roundTripped.array(), StandardCharsets.UTF_8));
+    }
+
+    @Test
+    void shouldThrowOnShortBuffer() throws NoSuchAlgorithmException {
+        var keygen = KeyGenerator.getInstance("AES");
+        var enc = AesGcmEncryptor.forEncrypt(new AesGcmIvGenerator(new SecureRandom()), keygen.generateKey());
+        var plaintext = "hello, world!";
+        byte[] plaintextBytes = plaintext.getBytes(StandardCharsets.UTF_8);
+        var size = enc.outputSize(plaintextBytes.length);
+        var ciphertext = ByteBuffer.allocate(size - 1);
+        ByteBuffer wrap = ByteBuffer.wrap(plaintextBytes);
+        var e = assertThrows(EncryptionException.class, () -> enc.encrypt(wrap, ciphertext));
+        assertInstanceOf(ShortBufferException.class, e.getCause());
+    }
+
+    @Test
+    void shouldThrowOnBadKey() {
+        var badKey = new SecretKeySpec(new byte[3], "AES");
+        var enc = AesGcmEncryptor.forEncrypt(new AesGcmIvGenerator(new SecureRandom()), badKey);
+        var plaintext = "hello, world!";
+        byte[] plaintextBytes = plaintext.getBytes(StandardCharsets.UTF_8);
+        var e = assertThrows(EncryptionException.class, () -> enc.outputSize(plaintextBytes.length));
+        assertInstanceOf(InvalidKeyException.class, e.getCause());
+    }
+
+    @Test
+    void shouldThrowDeserializingUnexpectedIvLength() throws NoSuchAlgorithmException {
+        var keygen = KeyGenerator.getInstance("AES");
+        var enc = AesGcmEncryptor.forDecrypt(keygen.generateKey());
+
+        ByteBuffer encoded = ByteBuffer.wrap(new byte[]{ 0, (byte) 1 });
+        ByteBuffer output = ByteBuffer.allocate(10);
+        var e = assertThrows(EncryptionException.class, () -> enc.decrypt(encoded, output));
+        assertEquals("Unexpected IV length", e.getMessage());
+    }
+
+    @Test
+    void shouldDetectChangedCiphertext() throws NoSuchAlgorithmException {
+        var keygen = KeyGenerator.getInstance("AES");
+        var enc = AesGcmEncryptor.forEncrypt(new AesGcmIvGenerator(new SecureRandom()), keygen.generateKey());
+        var plaintext = "hello, world!";
+        byte[] plaintextBytes = plaintext.getBytes(StandardCharsets.UTF_8);
+        var size = enc.outputSize(plaintextBytes.length);
+        var ciphertext = ByteBuffer.allocate(size);
+        enc.encrypt(ByteBuffer.wrap(plaintextBytes), ciphertext);
+        assertFalse(ciphertext.hasRemaining());
+        int pos = ciphertext.position() - 1;
+        ciphertext.put(pos, (byte) (ciphertext.get(pos) + 1));
+        ciphertext.flip();
+        var roundTripped = ByteBuffer.allocate(plaintextBytes.length);
+
+        var dec = AesGcmEncryptor.forDecrypt(keygen.generateKey());
+        var e = assertThrows(EncryptionException.class, () -> dec.decrypt(ciphertext, roundTripped));
+        assertInstanceOf(AEADBadTagException.class, e.getCause());
+
+    }
+
+    @Test
+    void shouldThrowOnUnknownVersion() throws NoSuchAlgorithmException {
+        var keygen = KeyGenerator.getInstance("AES");
+        var enc = AesGcmEncryptor.forEncrypt(new AesGcmIvGenerator(new SecureRandom()), keygen.generateKey());
+        var plaintext = "hello, world!";
+        byte[] plaintextBytes = plaintext.getBytes(StandardCharsets.UTF_8);
+        var size = enc.outputSize(plaintextBytes.length);
+        var ciphertext = ByteBuffer.allocate(size);
+        enc.encrypt(ByteBuffer.wrap(plaintextBytes), ciphertext);
+        assertFalse(ciphertext.hasRemaining());
+        ciphertext.put(0, Byte.MAX_VALUE);
+        ciphertext.flip();
+        var roundTripped = ByteBuffer.allocate(plaintextBytes.length);
+
+        var dec = AesGcmEncryptor.forDecrypt(keygen.generateKey());
+        var e = assertThrows(EncryptionException.class, () -> dec.decrypt(ciphertext, roundTripped));
+        assertEquals("Unknown version 127", e.getMessage());
+    }
+
+    @Test
+    void shouldByCopySafe() throws NoSuchAlgorithmException {
+        var keygen = KeyGenerator.getInstance("AES");
+        SecretKey key = keygen.generateKey();
+        var enc = AesGcmEncryptor.forEncrypt(new AesGcmIvGenerator(new SecureRandom()), key);
+
+        var str = "hello, world!";
+        var buffer = ByteBuffer.allocate(200);
+        var plaintext = buffer.duplicate();
+        plaintext.put(str.getBytes(StandardCharsets.UTF_8));
+        plaintext.flip();
+
+        var ciphertext = buffer.duplicate();
+        enc.outputSize(plaintext.limit());
+        enc.encrypt(plaintext, ciphertext);
+
+        ciphertext.flip();
+
+        var roundTripped = buffer.duplicate();
+        var dec = AesGcmEncryptor.forDecrypt(key);
+        dec.decrypt(ciphertext, roundTripped);
+        assertEquals(str, StandardCharsets.UTF_8.decode(roundTripped.flip()).toString());
+    }
+
+}
diff --git a/kroxylicious-filters/kroxylicious-encryption/src/test/java/io/kroxylicious/filter/encryption/inband/AesGcmIvGeneratorTest.java b/kroxylicious-filters/kroxylicious-encryption/src/test/java/io/kroxylicious/filter/encryption/inband/AesGcmIvGeneratorTest.java
new file mode 100644
index 00000000..d3336dd3
--- /dev/null
+++ b/kroxylicious-filters/kroxylicious-encryption/src/test/java/io/kroxylicious/filter/encryption/inband/AesGcmIvGeneratorTest.java
@@ -0,0 +1,108 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.filter.encryption.inband;
+
+import java.security.SecureRandom;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class AesGcmIvGeneratorTest {
+
+    @Test
+    void shouldHave12ByteSize() {
+        assertEquals(12, new AesGcmIvGenerator(new SecureRandom()).sizeBytes());
+    }
+
+    @Test
+    void shouldIncrement() {
+
+        var generator = new AesGcmIvGenerator(random(0, 0, 0));
+        byte[] iv = new byte[generator.sizeBytes()];
+        generator.generateIv(iv);
+        assertArrayEquals(new byte[]{ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0 }, iv);
+        generator.generateIv(iv);
+        assertArrayEquals(new byte[]{ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, (byte) 1 }, iv);
+        generator.generateIv(iv);
+        assertArrayEquals(new byte[]{ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, (byte) 2 }, iv);
+    }
+
+    @Test
+    void shouldCarry1() {
+        var generator = new AesGcmIvGenerator(random(Integer.MAX_VALUE, 0, 0));
+        byte[] iv = new byte[generator.sizeBytes()];
+        generator.generateIv(iv);
+        assertArrayEquals(new byte[]{ 0, 0, 0, 0, 0, 0, 0, 0, (byte) 127, (byte) -1, (byte) -1, (byte) -1 }, iv);
+        generator.generateIv(iv);
+        assertArrayEquals(new byte[]{ 0, 0, 0, 0, 0, 0, 0, 1, (byte) -128, 0, 0, 0 }, iv);
+        generator.generateIv(iv);
+        assertArrayEquals(new byte[]{ 0, 0, 0, 0, 0, 0, 0, 1, (byte) -128, 0, 0, (byte) 1 }, iv);
+    }
+
+    @Test
+    void shouldCarry2() {
+        var generator = new AesGcmIvGenerator(random(Integer.MAX_VALUE, Integer.MAX_VALUE, 0));
+        byte[] iv = new byte[generator.sizeBytes()];
+        generator.generateIv(iv);
+        assertArrayEquals(new byte[]{ 0, 0, 0, 0, (byte) 127, (byte) -1, (byte) -1, (byte) -1, (byte) 127, (byte) -1, (byte) -1, (byte) -1 }, iv);
+        generator.generateIv(iv);
+        assertArrayEquals(new byte[]{ 0, 0, 0, 1, (byte) -128, 0, 0, 0, (byte) -128, 0, 0, 0 }, iv);
+        generator.generateIv(iv);
+        assertArrayEquals(new byte[]{ 0, 0, 0, 1, (byte) -128, 0, 0, 0, (byte) -128, 0, 0, (byte) 1 }, iv);
+    }
+
+    @Test
+    void shouldCarry3() {
+        var generator = new AesGcmIvGenerator(random(Integer.MAX_VALUE, Integer.MAX_VALUE, Integer.MAX_VALUE));
+        byte[] iv = new byte[generator.sizeBytes()];
+        generator.generateIv(iv);
+        assertArrayEquals(
+                new byte[]{ (byte) 127, (byte) -1, (byte) -1, (byte) -1, (byte) 127, (byte) -1, (byte) -1, (byte) -1, (byte) 127, (byte) -1, (byte) -1, (byte) -1 }, iv);
+        generator.generateIv(iv);
+        assertArrayEquals(new byte[]{ (byte) -128, 0, 0, 0, (byte) -128, 0, 0, 0, (byte) -128, 0, 0, 0 }, iv);
+        generator.generateIv(iv);
+        assertArrayEquals(new byte[]{ (byte) -128, 0, 0, 0, (byte) -128, 0, 0, 0, (byte) -128, 0, 0, (byte) 1 }, iv);
+    }
+
+    private static SecureRandom random(int first, int second, int third) {
+        return new SecureRandom() {
+            private int count = 0;
+
+            @Override
+            public int nextInt() {
+                var count = this.count;
+                this.count++;
+                if (count == 0) {
+                    return first;
+                }
+                else if (count == 1) {
+                    return second;
+                }
+                else if (count == 2) {
+                    return third;
+                }
+                else {
+                    throw new IllegalStateException();
+                }
+            }
+        };
+    }
+
+    @Test
+    void shouldBeDestroyable() {
+
+        var generator = new AesGcmIvGenerator(random(0, 0, 0));
+        generator.destroy();
+        assertTrue(generator.isDestroyed());
+        assertThrows(IllegalStateException.class, () -> generator.generateIv(null));
+    }
+
+}
diff --git a/kroxylicious-filters/kroxylicious-encryption/src/test/java/io/kroxylicious/filter/encryption/inband/BufferPoolTest.java b/kroxylicious-filters/kroxylicious-encryption/src/test/java/io/kroxylicious/filter/encryption/inband/BufferPoolTest.java
new file mode 100644
index 00000000..d824ecfd
--- /dev/null
+++ b/kroxylicious-filters/kroxylicious-encryption/src/test/java/io/kroxylicious/filter/encryption/inband/BufferPoolTest.java
@@ -0,0 +1,23 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.filter.encryption.inband;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+class BufferPoolTest {
+
+    @Test
+    void testAllocating() {
+        BufferPool allocating = BufferPool.allocating();
+        var bb = allocating.acquire(123);
+        assertEquals(123, bb.capacity());
+        assertEquals(0, bb.position());
+        allocating.release(bb);
+    }
+}
\ No newline at end of file
diff --git a/kroxylicious-filters/kroxylicious-encryption/src/test/java/io/kroxylicious/filter/encryption/inband/InBandKeyManagerTest.java b/kroxylicious-filters/kroxylicious-encryption/src/test/java/io/kroxylicious/filter/encryption/inband/InBandKeyManagerTest.java
new file mode 100644
index 00000000..d6d47f7a
--- /dev/null
+++ b/kroxylicious-filters/kroxylicious-encryption/src/test/java/io/kroxylicious/filter/encryption/inband/InBandKeyManagerTest.java
@@ -0,0 +1,225 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.filter.encryption.inband;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.EnumSet;
+import java.util.List;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import javax.security.auth.DestroyFailedException;
+import javax.security.auth.Destroyable;
+
+import org.apache.kafka.common.header.Header;
+import org.apache.kafka.common.header.internals.RecordHeader;
+import org.junit.jupiter.api.Test;
+
+import io.kroxylicious.filter.encryption.EncryptionScheme;
+import io.kroxylicious.filter.encryption.Receiver;
+import io.kroxylicious.filter.encryption.RecordField;
+import io.kroxylicious.kms.provider.kroxylicious.inmemory.InMemoryKms;
+import io.kroxylicious.kms.provider.kroxylicious.inmemory.UnitTestingKmsService;
+
+import edu.umd.cs.findbugs.annotations.NonNull;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class InBandKeyManagerTest {
+
+    @Test
+    void shouldBeAbleToDependOnRecordHeaderEquality() {
+        // The InBandKeyManager relies internally on RecordHeader implementing equals
+        // Since it's Kafka's class let's validate that here
+        var rh = new RecordHeader("foo", new byte[]{ 7, 4, 1 });
+        var rh2 = new RecordHeader("foo", new byte[]{ 7, 4, 1 });
+        var rh3 = new RecordHeader("bar", new byte[]{ 3, 3 });
+
+        assertEquals(rh, rh2);
+        assertNotEquals(rh, rh3);
+        assertNotEquals(rh2, rh3);
+    }
+
+    @Test
+    void headers() {
+        var myHeader = new RecordHeader("mine", new byte[]{ 1 });
+        var nonEmpty = new Header[]{ myHeader };
+        var fieldsHeader = InBandKeyManager.createEncryptedFieldsHeader(EnumSet.of(RecordField.RECORD_VALUE));
+
+        var r1 = InBandKeyManager.prependToHeaders(new Header[0], fieldsHeader);
+        assertEquals(1, r1.length);
+        assertEquals(fieldsHeader, r1[0]);
+
+        var r2 = InBandKeyManager.prependToHeaders(nonEmpty, fieldsHeader);
+        assertEquals(2, r2.length);
+        assertEquals(fieldsHeader, r2[0]);
+        assertEquals(myHeader, r2[1]);
+
+        var x0 = InBandKeyManager.removeInitialHeaders(r1, 1);
+        assertEquals(0, x0.length);
+
+        var x1 = InBandKeyManager.removeInitialHeaders(r2, 1);
+        assertEquals(1, x1.length);
+        assertEquals(myHeader, x1[0]);
+    }
+
+    @Test
+    void testDestroy() {
+        AtomicBoolean called = new AtomicBoolean();
+        InBandKeyManager.destroy(new Destroyable() {
+            @Override
+            public void destroy() throws DestroyFailedException {
+                called.set(true);
+                throw new DestroyFailedException("Eeek!");
+            }
+        });
+
+        assertTrue(called.get());
+    }
+
+    @Test
+    void shouldEncryptRecordValue() {
+        var kmsService = UnitTestingKmsService.newInstance();
+        InMemoryKms kms = kmsService.buildKms(new UnitTestingKmsService.Config());
+        var km = new InBandKeyManager<>(kms, BufferPool.allocating());
+
+        var kekId = kms.generateKey();
+
+        var value = ByteBuffer.wrap(new byte[]{ 1, 2, 3 });
+        TestingRecord record = new TestingRecord(value);
+
+        List<TestingRecord> encrypted = new ArrayList<>();
+        List<TestingRecord> initial = List.of(record);
+        km.encrypt(new EncryptionScheme<>(kekId, EnumSet.of(RecordField.RECORD_VALUE)),
+                initial,
+                recordReceivedRecord(encrypted));
+        record.value().rewind();
+        assertEquals(1, encrypted.size());
+        assertNotEquals(initial, encrypted);
+        // TODO add assertion on headers
+
+        List<TestingRecord> decrypted = new ArrayList<>();
+        km.decrypt(encrypted, recordReceivedRecord(decrypted));
+
+        assertEquals(initial, decrypted);
+    }
+
+    @NonNull
+    private static Receiver recordReceivedRecord(Collection<TestingRecord> list) {
+        return (r, v, h) -> {
+            list.add(new TestingRecord(copyBytes(v), h));
+        };
+    }
+
+    @Test
+    void shouldEncryptRecordValueForMultipleRecords() throws ExecutionException, InterruptedException, TimeoutException {
+        var kmsService = UnitTestingKmsService.newInstance();
+        InMemoryKms kms = kmsService.buildKms(new UnitTestingKmsService.Config());
+        var km = new InBandKeyManager<>(kms, BufferPool.allocating());
+
+        var kekId = kms.generateKey();
+
+        var value = ByteBuffer.wrap(new byte[]{ 1, 2, 3 });
+        TestingRecord record = new TestingRecord(value);
+
+        var value2 = ByteBuffer.wrap(new byte[]{ 3, 4, 5 });
+        TestingRecord record2 = new TestingRecord(value2);
+
+        List<TestingRecord> encrypted = new ArrayList<>();
+        List<TestingRecord> initial = List.of(record, record2);
+        km.encrypt(new EncryptionScheme<>(kekId, EnumSet.of(RecordField.RECORD_VALUE)),
+                initial,
+                (r, v, h) -> {
+                    encrypted.add(new TestingRecord(copyBytes(v), h));
+                }).toCompletableFuture().get(10, TimeUnit.SECONDS);
+        record.value().rewind();
+        record2.value().rewind();
+        assertEquals(2, encrypted.size());
+        assertNotEquals(initial, encrypted);
+        // TODO add assertion on headers
+
+        List<TestingRecord> decrypted = new ArrayList<>();
+        km.decrypt(encrypted, recordReceivedRecord(decrypted)).toCompletableFuture().get(10, TimeUnit.SECONDS);
+
+        assertEquals(initial, decrypted);
+    }
+
+    @NonNull
+    private static ByteBuffer copyBytes(ByteBuffer v) {
+        byte[] bytes = new byte[v.remaining()];
+        v.get(bytes);
+        ByteBuffer wrap = ByteBuffer.wrap(bytes);
+        return wrap;
+    }
+
+    @Test
+    void shouldEncryptRecordHeaders() {
+        var kmsService = UnitTestingKmsService.newInstance();
+        InMemoryKms kms = kmsService.buildKms(new UnitTestingKmsService.Config());
+        var km = new InBandKeyManager<>(kms, BufferPool.allocating());
+
+        var kekId = kms.generateKey();
+
+        var value = ByteBuffer.wrap(new byte[]{ 1, 2, 3 });
+        var headers = new Header[]{ new RecordHeader("foo", new byte[]{ 4, 5, 6 }) };
+        TestingRecord record = new TestingRecord(value, headers);
+
+        List<TestingRecord> encrypted = new ArrayList<>();
+        List<TestingRecord> initial = List.of(record);
+        km.encrypt(new EncryptionScheme<>(kekId, EnumSet.of(RecordField.RECORD_HEADER_VALUES)),
+                initial,
+                recordReceivedRecord(encrypted));
+        value.rewind();
+
+        assertEquals(1, encrypted.size());
+        assertNotEquals(initial, encrypted);
+
+        List<TestingRecord> decrypted = new ArrayList<>();
+        km.decrypt(encrypted, recordReceivedRecord(decrypted));
+
+        assertEquals(List.of(new TestingRecord(value, new Header[]{ new RecordHeader("foo", new byte[]{ 4, 5, 6 }) })), decrypted);
+    }
+
+    @Test
+    void shouldEncryptRecordHeadersForMultipleRecords() throws ExecutionException, InterruptedException, TimeoutException {
+        var kmsService = UnitTestingKmsService.newInstance();
+        InMemoryKms kms = kmsService.buildKms(new UnitTestingKmsService.Config());
+        var km = new InBandKeyManager<>(kms, BufferPool.allocating());
+
+        var kekId = kms.generateKey();
+
+        var value = ByteBuffer.wrap(new byte[]{ 1, 2, 3 });
+        var headers = new Header[]{ new RecordHeader("foo", new byte[]{ 4, 5, 6 }) };
+        TestingRecord record = new TestingRecord(value, headers);
+        var value2 = ByteBuffer.wrap(new byte[]{ 7, 8, 9 });
+        var headers2 = new Header[]{ new RecordHeader("foo", new byte[]{ 10, 11, 12 }) };
+        TestingRecord record2 = new TestingRecord(value2, headers2);
+
+        List<TestingRecord> encrypted = new ArrayList<>();
+        List<TestingRecord> initial = List.of(record, record2);
+        km.encrypt(new EncryptionScheme<>(kekId, EnumSet.of(RecordField.RECORD_HEADER_VALUES)),
+                initial,
+                recordReceivedRecord(encrypted)).toCompletableFuture().get(10, TimeUnit.SECONDS);
+        value.rewind();
+        value2.rewind();
+        assertEquals(2, encrypted.size());
+        assertNotEquals(initial, encrypted);
+
+        List<TestingRecord> decrypted = new ArrayList<>();
+        km.decrypt(encrypted, recordReceivedRecord(decrypted));
+
+        assertEquals(List.of(new TestingRecord(value, new Header[]{ new RecordHeader("foo", new byte[]{ 4, 5, 6 }) }),
+                new TestingRecord(value2, new Header[]{ new RecordHeader("foo", new byte[]{ 10, 11, 12 }) })), decrypted);
+    }
+
+}
diff --git a/kroxylicious-filters/kroxylicious-encryption/src/test/java/io/kroxylicious/filter/encryption/inband/KeyContextTest.java b/kroxylicious-filters/kroxylicious-encryption/src/test/java/io/kroxylicious/filter/encryption/inband/KeyContextTest.java
new file mode 100644
index 00000000..0f51103b
--- /dev/null
+++ b/kroxylicious-filters/kroxylicious-encryption/src/test/java/io/kroxylicious/filter/encryption/inband/KeyContextTest.java
@@ -0,0 +1,90 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.filter.encryption.inband;
+
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.security.SecureRandom;
+import java.util.concurrent.ExecutionException;
+
+import javax.security.auth.DestroyFailedException;
+
+import org.junit.jupiter.api.Test;
+
+import io.kroxylicious.kms.provider.kroxylicious.inmemory.InMemoryKms;
+import io.kroxylicious.kms.provider.kroxylicious.inmemory.UnitTestingKmsService;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class KeyContextTest {
+
+    @Test
+    void shouldRejectInvalidNumberOfEncryptions() {
+        assertThrows(IllegalArgumentException.class, () -> new KeyContext(null, 101011, 0, null));
+        assertThrows(IllegalArgumentException.class, () -> new KeyContext(null, 101011, -1, null));
+    }
+
+    @Test
+    void shouldRejectInvalidRemainingOfEncryptions() throws ExecutionException, InterruptedException {
+        InMemoryKms kms = UnitTestingKmsService.newInstance().buildKms(new UnitTestingKmsService.Config());
+        var kek = kms.generateKey();
+        var pair = kms.generateDekPair(kek).get();
+        ByteBuffer prefix = ByteBuffer.wrap(new byte[]{ 1, 2, 3 });
+        var context = new KeyContext(prefix, 101011, 2,
+                AesGcmEncryptor.forEncrypt(new AesGcmIvGenerator(new SecureRandom()), pair.dek()));
+
+        assertThrows(IllegalArgumentException.class, () -> context.hasAtLeastRemainingEncryptions(0));
+        assertThrows(IllegalArgumentException.class, () -> context.hasAtLeastRemainingEncryptions(-1));
+    }
+
+    @Test
+    void testLifecycle() throws ExecutionException, InterruptedException, DestroyFailedException {
+        InMemoryKms kms = UnitTestingKmsService.newInstance().buildKms(new UnitTestingKmsService.Config());
+        var kek = kms.generateKey();
+        var pair = kms.generateDekPair(kek).get();
+        ByteBuffer prefix = ByteBuffer.wrap(new byte[]{ 1, 2, 3 });
+        var context = new KeyContext(prefix, 101011, 2,
+                AesGcmEncryptor.forEncrypt(new AesGcmIvGenerator(new SecureRandom()), pair.dek()));
+
+        assertTrue(context.hasAtLeastRemainingEncryptions(1));
+
+        ByteBuffer bb = ByteBuffer.wrap("hello, world!".getBytes(StandardCharsets.UTF_8));
+        int size = context.encodedSize(bb.capacity());
+        assertEquals(43, size);
+        var output = ByteBuffer.allocate(size);
+
+        // do an encryption
+        context.encode(bb, output);
+        assertFalse(output.hasRemaining());
+        output.flip();
+        bb.flip();
+
+        assertTrue(context.hasAtLeastRemainingEncryptions(1));
+        assertFalse(context.isExpiredForEncryption(101010));
+        assertFalse(context.isExpiredForEncryption(101011));
+        assertTrue(context.isExpiredForEncryption(101012));
+
+        // do a second encryption
+        context.encodedSize(bb.capacity());
+        context.encode(bb, output);
+        assertFalse(output.hasRemaining());
+        output.flip();
+        bb.flip();
+
+        assertFalse(context.hasAtLeastRemainingEncryptions(1));
+        context.encodedSize(bb.capacity());
+        // try to do a third encryption
+        var e = assertThrows(ExhaustedDekException.class, () -> context.encode(bb, output));
+        assertEquals("No more encryptions", e.getMessage());
+
+        // destroy
+        assertThrows(DestroyFailedException.class, context::destroy);
+    }
+}
diff --git a/kroxylicious-filters/kroxylicious-encryption/src/test/java/io/kroxylicious/filter/encryption/inband/TestingRecord.java b/kroxylicious-filters/kroxylicious-encryption/src/test/java/io/kroxylicious/filter/encryption/inband/TestingRecord.java
new file mode 100644
index 00000000..7253639c
--- /dev/null
+++ b/kroxylicious-filters/kroxylicious-encryption/src/test/java/io/kroxylicious/filter/encryption/inband/TestingRecord.java
@@ -0,0 +1,108 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.filter.encryption.inband;
+
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.Objects;
+
+import org.apache.kafka.common.header.Header;
+import org.apache.kafka.common.record.TimestampType;
+
+public record TestingRecord(ByteBuffer value, Header... headers) implements org.apache.kafka.common.record.Record {
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        TestingRecord that = (TestingRecord) o;
+        return Objects.equals(value, that.value) && Arrays.equals(headers, that.headers);
+    }
+
+    @Override
+    public int hashCode() {
+        int result = Objects.hash(value);
+        result = 31 * result + Arrays.hashCode(headers);
+        return result;
+    }
+
+    @Override
+    public String toString() {
+        return "TestingRecord{" +
+                "value=" + value +
+                ", headers=" + Arrays.toString(headers) +
+                '}';
+    }
+
+    @Override
+    public long offset() {
+        return 0;
+    }
+
+    @Override
+    public int sequence() {
+        return 0;
+    }
+
+    @Override
+    public int sizeInBytes() {
+        return 0;
+    }
+
+    @Override
+    public long timestamp() {
+        return 0;
+    }
+
+    @Override
+    public void ensureValid() {
+
+    }
+
+    @Override
+    public int keySize() {
+        return 0;
+    }
+
+    @Override
+    public boolean hasKey() {
+        return false;
+    }
+
+    @Override
+    public ByteBuffer key() {
+        return null;
+    }
+
+    @Override
+    public int valueSize() {
+        return value.limit();
+    }
+
+    @Override
+    public boolean hasValue() {
+        return value != null;
+    }
+
+    @Override
+    public boolean hasMagic(byte magic) {
+        return false;
+    }
+
+    @Override
+    public boolean isCompressed() {
+        return false;
+    }
+
+    @Override
+    public boolean hasTimestampType(TimestampType timestampType) {
+        return false;
+    }
+}
\ No newline at end of file
diff --git a/kroxylicious-filters/kroxylicious-encryption/src/test/resources/log4j2-test.properties b/kroxylicious-filters/kroxylicious-encryption/src/test/resources/log4j2-test.properties
new file mode 100644
index 00000000..a8857c02
--- /dev/null
+++ b/kroxylicious-filters/kroxylicious-encryption/src/test/resources/log4j2-test.properties
@@ -0,0 +1,21 @@
+#
+# Copyright Kroxylicious Authors.
+#
+# Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+#
+
+name = Config
+
+appender.null.type = Null
+appender.null.name = NullAppender
+
+appender.console.type = Console
+appender.console.name = STDOUT
+appender.console.layout.type = PatternLayout
+appender.console.layout.pattern = %d{yyyy-MM-dd HH:mm:ss} <%t> %-5p %c:%L - %m%n
+
+rootLogger.level = TRACE
+rootLogger.appenderRefs = null,console
+rootLogger.appenderRef.console.ref = STDOUT
+rootLogger.appenderRef.console.level = WARN
+rootLogger.additivity = false
diff --git a/kroxylicious-integration-tests/pom.xml b/kroxylicious-integration-tests/pom.xml
index 1121f57f..25c7dac1 100644
--- a/kroxylicious-integration-tests/pom.xml
+++ b/kroxylicious-integration-tests/pom.xml
@@ -38,6 +38,16 @@
             <artifactId>kroxylicious-integration-test-support</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>io.kroxylicious</groupId>
+            <artifactId>kroxylicious-encryption</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>io.kroxylicious</groupId>
+            <artifactId>kroxylicious-kms-provider-kroxylicious-inmemory</artifactId>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>io.kroxylicious</groupId>
             <artifactId>kroxylicious-multitenant</artifactId>
diff --git a/kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/encryption/EnvelopeEncryptionFilterIT.java b/kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/encryption/EnvelopeEncryptionFilterIT.java
new file mode 100644
index 00000000..81b68719
--- /dev/null
+++ b/kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/encryption/EnvelopeEncryptionFilterIT.java
@@ -0,0 +1,356 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.proxy.encryption;
+
+import java.time.Duration;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
+
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.TopicPartition;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import io.kroxylicious.filter.encryption.TemplateKekSelector;
+import io.kroxylicious.kms.provider.kroxylicious.inmemory.InMemoryKms;
+import io.kroxylicious.kms.provider.kroxylicious.inmemory.IntegrationTestingKmsService;
+import io.kroxylicious.proxy.config.FilterDefinition;
+import io.kroxylicious.proxy.config.FilterDefinitionBuilder;
+import io.kroxylicious.testing.kafka.api.KafkaCluster;
+import io.kroxylicious.testing.kafka.junit5ext.KafkaClusterExtension;
+
+import edu.umd.cs.findbugs.annotations.NonNull;
+
+import static io.kroxylicious.test.tester.KroxyliciousConfigUtils.DEFAULT_VIRTUAL_CLUSTER;
+import static io.kroxylicious.test.tester.KroxyliciousConfigUtils.proxy;
+import static io.kroxylicious.test.tester.KroxyliciousTesters.kroxyliciousTester;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.testcontainers.shaded.org.awaitility.Awaitility.await;
+
+@ExtendWith(KafkaClusterExtension.class)
+class EnvelopeEncryptionFilterIT {
+
+    @Test
+    void roundTrip(KafkaCluster cluster) throws Exception {
+        var kmsId = UUID.randomUUID();
+        var kms = provisionKms(kmsId);
+        generateKekWithAlias(kms);
+
+        var builder = proxy(cluster);
+
+        builder.addToFilters(encryptionFilterDefinition(kmsId));
+
+        try (var tester = kroxyliciousTester(builder);
+                var admin = tester.admin();
+                var producer = tester.producer();
+                var consumer = tester.consumer()) {
+
+            String topic = tester.createTopic(DEFAULT_VIRTUAL_CLUSTER);
+
+            await().atMost(Duration.ofSeconds(5)).until(() -> admin.listTopics().namesToListings().get(),
+                    n -> n.containsKey(topic));
+
+            var message = "hello world";
+            producer.send(new ProducerRecord<>(topic, message)).get(5, TimeUnit.SECONDS);
+
+            consumer.subscribe(List.of(topic));
+            var records = consumer.poll(Duration.ofSeconds(2));
+            assertThat(records.iterator())
+                    .toIterable()
+                    .singleElement()
+                    .extracting(ConsumerRecord::value)
+                    .isEqualTo(message);
+        }
+    }
+
+    // This ensures the decrypt-ability guarantee, post kek rotation
+    @Test
+    void decryptionAfterKekRotation(KafkaCluster cluster, Admin admin) throws Exception {
+        var kmsId = UUID.randomUUID();
+        var kms = provisionKms(kmsId);
+        generateKekWithAlias(kms);
+
+        var topicName = UUID.randomUUID().toString();
+        admin.createTopics(List.of(new NewTopic(topicName, Optional.empty(), Optional.empty()))).all().get(5, TimeUnit.SECONDS);
+        await().atMost(Duration.ofSeconds(5)).until(() -> admin.listTopics().namesToListings().get(),
+                n -> n.containsKey(topicName));
+
+        var builder = proxy(cluster);
+        builder.addToFilters(encryptionFilterDefinition(kmsId));
+
+        var messageBeforeKeyRotation = "hello world, old key";
+        var messageAfterKeyRotation = "hello world, new key";
+        try (var tester = kroxyliciousTester(builder);
+                var producer = tester.producer()) {
+            producer.send(new ProducerRecord<>(topicName, messageBeforeKeyRotation)).get(5, TimeUnit.SECONDS);
+        }
+
+        // Now do the Kek rotation
+        generateKekWithAlias(kms);
+
+        builder.addToFilters(0, encryptionFilterDefinition(kmsId));
+
+        try (var tester = kroxyliciousTester(builder);
+                var producer = tester.producer();
+                var consumer = tester.consumer()) {
+
+            producer.send(new ProducerRecord<>(topicName, messageAfterKeyRotation)).get(5, TimeUnit.SECONDS);
+
+            consumer.subscribe(List.of(topicName));
+            var records = consumer.poll(Duration.ofSeconds(2));
+            assertThat(records.iterator())
+                    .toIterable()
+                    .extracting(ConsumerRecord::value)
+                    .containsExactly(messageBeforeKeyRotation, messageAfterKeyRotation);
+        }
+
+        IntegrationTestingKmsService.delete(kmsId.toString());
+    }
+
+    @NonNull
+    private InMemoryKms provisionKms(UUID kmsId) {
+        return IntegrationTestingKmsService.newInstance().buildKms(new IntegrationTestingKmsService.Config(kmsId.toString()));
+    }
+
+    private UUID generateKekWithAlias(InMemoryKms kms) {
+        var kekId = kms.generateKey();
+        kms.createAlias(kekId, "all");
+        return kekId;
+    }
+
+    @Test
+    void topicRecordsAreUnreadableOnServer(KafkaCluster cluster, KafkaConsumer<String, String> directConsumer) throws Exception {
+        var kmsId = UUID.randomUUID();
+        var kms = provisionKms(kmsId);
+        generateKekWithAlias(kms);
+
+        var builder = proxy(cluster);
+        builder.addToFilters(encryptionFilterDefinition(kmsId));
+
+        try (var tester = kroxyliciousTester(builder);
+                var admin = tester.admin();
+                var producer = tester.producer()) {
+
+            String topic = tester.createTopic(DEFAULT_VIRTUAL_CLUSTER);
+
+            await().atMost(Duration.ofSeconds(5)).until(() -> admin.listTopics().namesToListings().get(),
+                    n -> n.containsKey(topic));
+
+            var message = "hello world";
+            producer.send(new ProducerRecord<>(topic, message)).get(5, TimeUnit.SECONDS);
+
+            var tps = List.of(new TopicPartition(topic, 0));
+            directConsumer.assign(tps);
+            directConsumer.seekToBeginning(tps);
+            var records = directConsumer.poll(Duration.ofSeconds(2));
+            assertThat(records.iterator())
+                    .toIterable()
+                    .singleElement()
+                    .extracting(ConsumerRecord::value)
+                    .isNotEqualTo(message);
+        }
+
+        IntegrationTestingKmsService.delete(kmsId.toString());
+    }
+
+    @Test
+    void unencryptedRecordsConsumable(KafkaCluster cluster, KafkaProducer<String, String> directProducer) throws Exception {
+        var kmsId = UUID.randomUUID();
+        var kms = provisionKms(kmsId);
+        generateKekWithAlias(kms);
+
+        var builder = proxy(cluster);
+
+        builder.addToFilters(encryptionFilterDefinition(kmsId));
+
+        try (var tester = kroxyliciousTester(builder);
+                var admin = tester.admin();
+                var producer = tester.producer();
+                var consumer = tester.consumer()) {
+
+            String topic = tester.createTopic(DEFAULT_VIRTUAL_CLUSTER);
+
+            await().atMost(Duration.ofSeconds(5)).until(() -> admin.listTopics().namesToListings().get(),
+                    n -> n.containsKey(topic));
+
+            // messages produced via Kroxylicious will be encrypted
+            var message = "hello encrypted world";
+            producer.send(new ProducerRecord<>(topic, message)).get(5, TimeUnit.SECONDS);
+
+            // messages produced direct will be plain
+            var plainMessage = "hello plain world";
+            directProducer.send(new ProducerRecord<>(topic, plainMessage)).get(5, TimeUnit.SECONDS);
+
+            consumer.subscribe(List.of(topic));
+            var records = consumer.poll(Duration.ofSeconds(2));
+            assertThat(records.iterator()).toIterable()
+                    .hasSize(2)
+                    .map(ConsumerRecord::value)
+                    .containsExactly(message, plainMessage);
+        }
+
+        IntegrationTestingKmsService.delete(kmsId.toString());
+    }
+
+    @Test
+    @Disabled("InBandKeyManger doesn't handle nulls")
+    void nullValueRecordProducedAndConsumedSuccessfully(KafkaCluster cluster) throws Exception {
+        var kmsId = UUID.randomUUID();
+        var kms = provisionKms(kmsId);
+        generateKekWithAlias(kms);
+
+        var builder = proxy(cluster);
+
+        builder.addToFilters(encryptionFilterDefinition(kmsId));
+
+        try (var tester = kroxyliciousTester(builder);
+                var admin = tester.admin();
+                var producer = tester.producer();
+                var consumer = tester.consumer()) {
+
+            String topic = tester.createTopic(DEFAULT_VIRTUAL_CLUSTER);
+
+            await().atMost(Duration.ofSeconds(5)).until(() -> admin.listTopics().namesToListings().get(),
+                    n -> n.containsKey(topic));
+
+            String message = null;
+            producer.send(new ProducerRecord<>(topic, message)).get(5, TimeUnit.SECONDS);
+
+            consumer.subscribe(List.of(topic));
+            var records = consumer.poll(Duration.ofSeconds(2));
+            assertThat(records.iterator())
+                    .toIterable()
+                    .singleElement()
+                    .extracting(ConsumerRecord::value)
+                    .isNull();
+        }
+
+        IntegrationTestingKmsService.delete(kmsId.toString());
+    }
+
+    @Test
+    void produceAndConsumeEncryptedAndPlainTopicsAtSameTime(KafkaCluster cluster, Admin admin) throws Exception {
+        var kmsId = UUID.randomUUID();
+        var kms = provisionKms(kmsId);
+        var keyId = generateKekWithAlias(kms);
+
+        var encryptedTopic = UUID.randomUUID().toString();
+        var plainTopic = UUID.randomUUID().toString();
+
+        var topics = Stream.of(encryptedTopic, plainTopic)
+                .map(t -> new NewTopic(t, Optional.empty(), Optional.empty()))
+                .toList();
+        admin.createTopics(topics).all().get(5, TimeUnit.SECONDS);
+        await().atMost(Duration.ofSeconds(5)).until(() -> admin.listTopics().namesToListings().get(),
+                n -> n.containsKey(encryptedTopic) && n.containsKey(plainTopic));
+
+        var builder = proxy(cluster);
+
+        builder.addToFilters(encryptionFilterDefinition(kmsId));
+
+        try (var tester = kroxyliciousTester(builder);
+                var producer = tester.producer(Map.of(ProducerConfig.LINGER_MS_CONFIG, 1000, ProducerConfig.BATCH_SIZE_CONFIG, 2));
+                var consumer = tester.consumer()) {
+
+            var secretMessage = "hello secret";
+            var plainMessage = "hello world";
+
+            producer.send(new ProducerRecord<>(encryptedTopic, secretMessage));
+            producer.send(new ProducerRecord<>(plainTopic, plainMessage));
+            producer.flush();
+
+            consumer.subscribe(List.of(encryptedTopic, plainTopic));
+            var records = consumer.poll(Duration.ofSeconds(2));
+            assertThat(records.iterator())
+                    .toIterable()
+                    .extracting(ConsumerRecord::value)
+                    .contains(secretMessage, plainMessage);
+        }
+
+        IntegrationTestingKmsService.delete(kmsId.toString());
+    }
+
+    @Test
+    void shouldGenerateOneDek(KafkaCluster cluster) throws Exception {
+        var kmsId = UUID.randomUUID();
+        var kms = provisionKms(kmsId);
+        var keyId = generateKekWithAlias(kms);
+
+        var builder = proxy(cluster);
+
+        builder.addToFilters(encryptionFilterDefinition(kmsId));
+
+        try (var tester = kroxyliciousTester(builder);
+                var admin = tester.admin();
+                var producer = tester.producer(Map.of(ProducerConfig.LINGER_MS_CONFIG, 0));
+                var consumer = tester.consumer()) {
+
+            String topic = tester.createTopic(DEFAULT_VIRTUAL_CLUSTER);
+
+            await().atMost(Duration.ofSeconds(5)).until(() -> admin.listTopics().namesToListings().get(),
+                    n -> n.containsKey(topic));
+
+            var message = "hello world";
+            producer.send(new ProducerRecord<>(topic, message)).get(5, TimeUnit.SECONDS);
+
+            consumer.subscribe(List.of(topic));
+            var records = consumer.poll(Duration.ofSeconds(2));
+            assertThat(records.iterator())
+                    .toIterable()
+                    .singleElement()
+                    .extracting(ConsumerRecord::value)
+                    .isEqualTo(message);
+
+            // Send two batches to the same topic
+            message = "hello world #2";
+            producer.send(new ProducerRecord<>(topic, message)).get(5, TimeUnit.SECONDS);
+
+            consumer.subscribe(List.of(topic));
+            records = consumer.poll(Duration.ofSeconds(2));
+            assertThat(records.iterator())
+                    .toIterable()
+                    .singleElement()
+                    .extracting(ConsumerRecord::value)
+                    .isEqualTo(message);
+        }
+
+        // We expect that only one DEK was generated
+        assertThat(kms.numDeksGenerated()).isEqualTo(1);
+
+        IntegrationTestingKmsService.delete(kmsId.toString());
+    }
+
+    /*
+     *
+     * further IT ideas:
+     * fetching from > 1 topics (mixed encryption/plain case)
+     * exploratory test examining what the client will see/do when decryption fails - looking to verify
+     * - behaviour is reasonable
+     * - the user has a chance to understand what's wrong.
+     * behaviour when KMS throws a KmsException a) during produce, b) during fetch
+     */
+    private FilterDefinition encryptionFilterDefinition(UUID kmsId) {
+        return new FilterDefinitionBuilder("EnvelopeEncryption")
+                .withConfig("kms", IntegrationTestingKmsService.class.getSimpleName())
+                .withConfig("kmsConfig", Map.of("name", kmsId.toString()))
+                .withConfig("selector", TemplateKekSelector.class.getSimpleName())
+                .withConfig("selectorConfig", Map.of("template", "all"))
+                .build();
+
+    }
+}