Repository: kroxylicious/kroxylicious
Pull request #632 (2023-09-27 04:03:59)

Topic manager
### 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).


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=632)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=632&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=632&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=632&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=632&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=632&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=632&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=632&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=632&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=632&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=632&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=632&resolved=false&types=CODE_SMELL) [0 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=632&resolved=false&types=CODE_SMELL)

[![52.2%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/50-16px.png '52.2%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=632&metric=new_coverage&view=list) [52.2% Coverage](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=632&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=632&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=632&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: I like the idea of the topic generator.
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=632)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=632&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=632&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=632&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=632&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=632&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=632&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=632&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=632&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=632&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=632&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=632&resolved=false&types=CODE_SMELL) [4 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=632&resolved=false&types=CODE_SMELL)

[![75.7%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/60-16px.png '75.7%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=632&metric=new_coverage&view=list) [75.7% Coverage](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=632&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=632&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=632&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]: 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=632)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=632&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=632&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=632&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=632&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=632&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=632&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=632&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=632&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=632&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=632&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=632&resolved=false&types=CODE_SMELL) [5 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=632&resolved=false&types=CODE_SMELL)

[![83.3%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/60-16px.png '83.3%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=632&metric=new_coverage&view=list) [83.3% Coverage](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=632&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=632&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=632&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=632)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=632&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=632&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=632&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=632&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=632&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=632&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=632&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=632&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=632&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=632&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=632&resolved=false&types=CODE_SMELL) [5 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=632&resolved=false&types=CODE_SMELL)

[![84.0%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/60-16px.png '84.0%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=632&metric=new_coverage&view=list) [84.0% Coverage](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=632&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=632&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=632&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=632)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=632&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=632&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=632&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=632&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=632&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=632&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=632&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=632&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=632&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=632&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=632&resolved=false&types=CODE_SMELL) [5 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=632&resolved=false&types=CODE_SMELL)

[![84.5%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/60-16px.png '84.5%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=632&metric=new_coverage&view=list) [84.5% Coverage](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=632&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=632&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=632&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=632)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=632&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=632&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=632&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=632&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=632&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=632&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=632&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=632&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=632&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=632&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=632&resolved=false&types=CODE_SMELL) [5 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=632&resolved=false&types=CODE_SMELL)

[![84.5%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/60-16px.png '84.5%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=632&metric=new_coverage&view=list) [84.5% Coverage](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=632&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=632&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=632&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=632)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=632&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=632&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=632&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=632&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=632&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=632&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=632&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=632&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=632&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=632&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=632&resolved=false&types=CODE_SMELL) [5 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=632&resolved=false&types=CODE_SMELL)

[![84.5%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/60-16px.png '84.5%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=632&metric=new_coverage&view=list) [84.5% Coverage](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=632&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=632&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=632&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=632)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=632&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=632&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=632&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=632&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=632&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=632&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=632&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=632&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=632&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=632&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=632&resolved=false&types=CODE_SMELL) [5 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=632&resolved=false&types=CODE_SMELL)

[![84.5%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/60-16px.png '84.5%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=632&metric=new_coverage&view=list) [84.5% Coverage](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=632&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=632&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=632&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=632)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=632&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=632&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=632&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=632&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=632&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=632&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=632&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=632&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=632&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=632&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=632&resolved=false&types=CODE_SMELL) [5 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=632&resolved=false&types=CODE_SMELL)

[![84.6%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/60-16px.png '84.6%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=632&metric=new_coverage&view=list) [84.6% Coverage](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=632&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=632&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=632&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=632)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=632&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=632&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=632&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=632&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=632&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=632&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=632&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=632&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=632&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=632&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=632&resolved=false&types=CODE_SMELL) [5 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=632&resolved=false&types=CODE_SMELL)

[![84.6%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/60-16px.png '84.6%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=632&metric=new_coverage&view=list) [84.6% Coverage](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=632&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=632&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=632&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=632)

[![Bug](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug-16px.png 'Bug')](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=632&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=632&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=632&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=632&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=632&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=632&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=632&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=632&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kroxylicious_kroxylicious&pullRequest=632&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=632&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=632&resolved=false&types=CODE_SMELL) [5 Code Smells](https://sonarcloud.io/project/issues?id=kroxylicious_kroxylicious&pullRequest=632&resolved=false&types=CODE_SMELL)

[![84.6%](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/60-16px.png '84.6%')](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=632&metric=new_coverage&view=list) [84.6% Coverage](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=632&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=632&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=kroxylicious_kroxylicious&pullRequest=632&metric=new_duplicated_lines_density&view=list)



Patch:
diff --git a/.idea/misc.xml b/.idea/misc.xml
index 1dd8bf6f..c7a2c708 100644
--- a/.idea/misc.xml
+++ b/.idea/misc.xml
@@ -8,6 +8,54 @@
       </list>
     </option>
   </component>
+  <component name="NullableNotNullManager">
+    <option name="myDefaultNullable" value="edu.umd.cs.findbugs.annotations.Nullable" />
+    <option name="myDefaultNotNull" value="edu.umd.cs.findbugs.annotations.NonNull" />
+    <option name="myNullables">
+      <value>
+        <list size="16">
+          <item index="0" class="java.lang.String" itemvalue="android.support.annotation.Nullable" />
+          <item index="1" class="java.lang.String" itemvalue="androidx.annotation.Nullable" />
+          <item index="2" class="java.lang.String" itemvalue="androidx.annotation.RecentlyNullable" />
+          <item index="3" class="java.lang.String" itemvalue="com.android.annotations.Nullable" />
+          <item index="4" class="java.lang.String" itemvalue="edu.umd.cs.findbugs.annotations.Nullable" />
+          <item index="5" class="java.lang.String" itemvalue="io.reactivex.annotations.Nullable" />
+          <item index="6" class="java.lang.String" itemvalue="io.reactivex.rxjava3.annotations.Nullable" />
+          <item index="7" class="java.lang.String" itemvalue="jakarta.annotation.Nullable" />
+          <item index="8" class="java.lang.String" itemvalue="javax.annotation.CheckForNull" />
+          <item index="9" class="java.lang.String" itemvalue="javax.annotation.Nullable" />
+          <item index="10" class="java.lang.String" itemvalue="org.checkerframework.checker.nullness.compatqual.NullableDecl" />
+          <item index="11" class="java.lang.String" itemvalue="org.checkerframework.checker.nullness.compatqual.NullableType" />
+          <item index="12" class="java.lang.String" itemvalue="org.checkerframework.checker.nullness.qual.Nullable" />
+          <item index="13" class="java.lang.String" itemvalue="org.eclipse.jdt.annotation.Nullable" />
+          <item index="14" class="java.lang.String" itemvalue="org.jetbrains.annotations.Nullable" />
+          <item index="15" class="java.lang.String" itemvalue="org.jspecify.nullness.Nullable" />
+        </list>
+      </value>
+    </option>
+    <option name="myNotNulls">
+      <value>
+        <list size="16">
+          <item index="0" class="java.lang.String" itemvalue="android.support.annotation.NonNull" />
+          <item index="1" class="java.lang.String" itemvalue="androidx.annotation.NonNull" />
+          <item index="2" class="java.lang.String" itemvalue="androidx.annotation.RecentlyNonNull" />
+          <item index="3" class="java.lang.String" itemvalue="com.android.annotations.NonNull" />
+          <item index="4" class="java.lang.String" itemvalue="edu.umd.cs.findbugs.annotations.NonNull" />
+          <item index="5" class="java.lang.String" itemvalue="io.reactivex.annotations.NonNull" />
+          <item index="6" class="java.lang.String" itemvalue="io.reactivex.rxjava3.annotations.NonNull" />
+          <item index="7" class="java.lang.String" itemvalue="jakarta.annotation.Nonnull" />
+          <item index="8" class="java.lang.String" itemvalue="javax.annotation.Nonnull" />
+          <item index="9" class="java.lang.String" itemvalue="lombok.NonNull" />
+          <item index="10" class="java.lang.String" itemvalue="org.checkerframework.checker.nullness.compatqual.NonNullDecl" />
+          <item index="11" class="java.lang.String" itemvalue="org.checkerframework.checker.nullness.compatqual.NonNullType" />
+          <item index="12" class="java.lang.String" itemvalue="org.checkerframework.checker.nullness.qual.NonNull" />
+          <item index="13" class="java.lang.String" itemvalue="org.eclipse.jdt.annotation.NonNull" />
+          <item index="14" class="java.lang.String" itemvalue="org.jetbrains.annotations.NotNull" />
+          <item index="15" class="java.lang.String" itemvalue="org.jspecify.nullness.NonNull" />
+        </list>
+      </value>
+    </option>
+  </component>
   <component name="PDMPlugin">
     <option name="skipTestSources" value="false" />
   </component>
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 3afc0135..6ff0b26e 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -6,6 +6,8 @@ Please enumerate **all user-facing** changes using format `<githib issue/pr numb
 
 ## 0.3.0
 
+* [#632](https://github.com/kroxylicious/kroxylicious/pull/632): Kroxylicious tester now supports creating & deleting topics on specific virtual clusters.
+* [#675](https://github.com/kroxylicious/kroxylicious/pull/675): Bump to Netty 4.1.100.Final to mitigate the Rapid Reset Attack (CVE-2023-44487)
 * [#665](https://github.com/kroxylicious/kroxylicious/pull/665): Bump org.apache.kafka:kafka-clients from 3.5.1 to 3.6.0
 * [#660](https://github.com/kroxylicious/kroxylicious/pull/660): Use container registry neutral terminology in docs/scripts #660
 * [#648](https://github.com/kroxylicious/kroxylicious/pull/648): Bump io.netty.incubator:netty-incubator-transport-native-io_uring from 0.0.22.Final to 0.0.23.Final
@@ -81,3 +83,6 @@ The names used to identify port-per-broker and sni-routing schemes in the Kroxyl
 The names used to identify micrometer configuration hooks in configuration have changed:
 - `CommonTagsContributor` -> `CommonTagsHook`
 - `StandardBindersContributor` -> `StandardBindersHook`
+- 
+#### CVE Fixes
+CVE-2023-44487 [#675](https://github.com/kroxylicious/kroxylicious/pull/675)
\ No newline at end of file
diff --git a/pom.xml b/pom.xml
index e779348a..3dc948fa 100644
--- a/pom.xml
+++ b/pom.xml
@@ -53,7 +53,7 @@
         <sundr-builder-annotations.version>0.101.0</sundr-builder-annotations.version>
         <spotbugs-annotations.version>4.7.3</spotbugs-annotations.version>
         <zjsonpatch.version>0.4.14</zjsonpatch.version>
-
+        <moby-names-generator.version>20.10.0-r0</moby-names-generator.version>
         <jacoco-maven-plugin.version>0.8.10</jacoco-maven-plugin.version>
     </properties>
 
@@ -162,6 +162,11 @@
                 <artifactId>picocli</artifactId>
                 <version>${picocli.version}</version>
             </dependency>
+            <dependency>
+                <groupId>info.schnatterer.moby-names-generator</groupId>
+                <artifactId>moby-names-generator</artifactId>
+                <version>${moby-names-generator.version}</version>
+            </dependency>
             <dependency>
                 <groupId>io.micrometer</groupId>
                 <artifactId>micrometer-bom</artifactId>


Test patch:
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/multitenant/BaseMultiTenantIT.java b/integrationtests/src/test/java/io/kroxylicious/proxy/multitenant/BaseMultiTenantIT.java
index 610b2ad5..08d28a98 100644
--- a/integrationtests/src/test/java/io/kroxylicious/proxy/multitenant/BaseMultiTenantIT.java
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/multitenant/BaseMultiTenantIT.java
@@ -81,7 +81,7 @@ public abstract class BaseMultiTenantIT extends BaseIT {
         this.certificateGenerator = new KeytoolCertificateGenerator();
         this.certificateGenerator.generateSelfSignedCertificateEntry("test@redhat.com", IntegrationTestInetAddressResolverProvider.generateFullyQualifiedDomainName("*"),
                 "KI", "RedHat", null, null, "US");
-        Path clientTrustStore = certsDirectory.resolve("kafka.truststore.jks");
+        Path clientTrustStore = certsDirectory.resolve(certificateGenerator.getTrustStoreLocation());
         this.certificateGenerator.generateTrustStore(this.certificateGenerator.getCertFilePath(), "client", clientTrustStore.toAbsolutePath().toString());
         this.clientConfig = Map.of(CommonClientConfigs.CLIENT_ID_CONFIG, testInfo.getDisplayName(),
                 CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, SecurityProtocol.SSL.name,
@@ -133,7 +133,7 @@ public abstract class BaseMultiTenantIT extends BaseIT {
         return getConsumerWithConfig(tester, Optional.of(virtualCluster), baseConfig, standardConfig, additionalConfig);
     }
 
-    void consumeAndVerify(KroxyliciousTester tester, Map<String, Object> clientConfig, String virtualCluster, String topicName, String groupId,
+    void consumeAndAssert(KroxyliciousTester tester, Map<String, Object> clientConfig, String virtualCluster, String topicName, String groupId,
                           Deque<Predicate<ConsumerRecord<String, String>>> expected, boolean offsetCommit) {
         try (var consumer = getConsumerWithConfig(tester, virtualCluster, groupId, clientConfig, Map.of(
                 ConsumerConfig.MAX_POLL_RECORDS_CONFIG, String.format("%d", expected.size())))) {
@@ -157,7 +157,7 @@ public abstract class BaseMultiTenantIT extends BaseIT {
         }
     }
 
-    void produceAndVerify(KroxyliciousTester tester, Map<String, Object> clientConfig, String virtualCluster,
+    void produceAndAssert(KroxyliciousTester tester, Map<String, Object> clientConfig, String virtualCluster,
                           Stream<ProducerRecord<String, String>> records, Optional<String> transactionalId) {
 
         Map<String, Object> config = new HashMap<>();
@@ -209,4 +209,4 @@ public abstract class BaseMultiTenantIT extends BaseIT {
             }
         };
     }
-}
+}
\ No newline at end of file
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/multitenant/MultiTenantIT.java b/integrationtests/src/test/java/io/kroxylicious/proxy/multitenant/MultiTenantIT.java
index efbe87ef..20087416 100644
--- a/integrationtests/src/test/java/io/kroxylicious/proxy/multitenant/MultiTenantIT.java
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/multitenant/MultiTenantIT.java
@@ -49,7 +49,9 @@ import org.junit.jupiter.params.provider.EnumSource;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import io.kroxylicious.proxy.config.ConfigurationBuilder;
 import io.kroxylicious.test.tester.KroxyliciousTester;
+import io.kroxylicious.test.tester.KroxyliciousTesters;
 import io.kroxylicious.testing.kafka.api.KafkaCluster;
 import io.kroxylicious.testing.kafka.junit5ext.KafkaClusterExtension;
 
@@ -106,7 +108,7 @@ class MultiTenantIT extends BaseMultiTenantIT {
     }
 
     @Test
-    void describeTopic(KafkaCluster cluster) throws Exception {
+    void describeTopic(KafkaCluster cluster) {
         var config = getConfig(cluster, this.certificateGenerator);
         try (var tester = kroxyliciousTester(config);
                 var admin = tester.admin(TENANT_1_CLUSTER, this.clientConfig)) {
@@ -122,71 +124,76 @@ class MultiTenantIT extends BaseMultiTenantIT {
     }
 
     @Test
-    void produceOne(KafkaCluster cluster) throws Exception {
+    void produceOne(KafkaCluster cluster) {
         var config = getConfig(cluster, this.certificateGenerator);
-        try (var tester = kroxyliciousTester(config);
-                var admin = tester.admin(TENANT_1_CLUSTER, this.clientConfig)) {
-            createTopics(admin, NEW_TOPIC_1);
-            produceAndVerify(tester, this.clientConfig, TENANT_1_CLUSTER, Stream.of(new ProducerRecord<>(TOPIC_1, MY_KEY, MY_VALUE)), Optional.empty());
+        try (var tester = buildTester(config)) {
+            final String topicName = tester.createTopic(TENANT_1_CLUSTER);
+            produceAndAssert(tester, this.clientConfig, TENANT_1_CLUSTER, Stream.of(new ProducerRecord<>(topicName, MY_KEY, MY_VALUE)), Optional.empty());
         }
     }
 
+    private KroxyliciousTester buildTester(ConfigurationBuilder config) {
+        return KroxyliciousTesters.newBuilder(config)
+                .setTrustStoreLocation(this.certificateGenerator.getTrustStoreLocation())
+                .setTrustStorePassword(this.certificateGenerator.getPassword())
+                .createDefaultKroxyliciousTester();
+    }
+
     @Test
-    void consumeOne(KafkaCluster cluster) throws Exception {
+    void consumeOne(KafkaCluster cluster) {
         var config = getConfig(cluster, this.certificateGenerator);
-        try (var tester = kroxyliciousTester(config);
-                var admin = tester.admin(TENANT_1_CLUSTER, this.clientConfig)) {
+        try (var tester = buildTester(config)) {
             var groupId = testInfo.getDisplayName();
-            createTopics(admin, NEW_TOPIC_1);
-            produceAndVerify(tester, this.clientConfig, TENANT_1_CLUSTER, Stream.of(new ProducerRecord<>(TOPIC_1, MY_KEY, MY_VALUE)), Optional.empty());
-            consumeAndVerify(tester, this.clientConfig, TENANT_1_CLUSTER, TOPIC_1, groupId, new LinkedList<>(List.of(matchesRecord(TOPIC_1, MY_KEY, MY_VALUE))), false);
+            final String topicName = tester.createTopic(TENANT_1_CLUSTER);
+            produceAndAssert(tester, this.clientConfig, TENANT_1_CLUSTER, Stream.of(new ProducerRecord<>(topicName, MY_KEY, MY_VALUE)), Optional.empty());
+            consumeAndAssert(tester, this.clientConfig, TENANT_1_CLUSTER, topicName, groupId, new LinkedList<>(List.of(matchesRecord(topicName, MY_KEY, MY_VALUE))),
+                    false);
         }
     }
 
     @Test
-    void consumeOneAndOffsetCommit(KafkaCluster cluster) throws Exception {
+    void consumeOneAndOffsetCommit(KafkaCluster cluster) {
         var config = getConfig(cluster, this.certificateGenerator);
-        try (var tester = kroxyliciousTester(config);
-                var admin = tester.admin(TENANT_1_CLUSTER, this.clientConfig)) {
+        try (var tester = buildTester(config)) {
             var groupId = testInfo.getDisplayName();
-            createTopics(admin, NEW_TOPIC_1);
-            produceAndVerify(tester, this.clientConfig, TENANT_1_CLUSTER,
-                    Stream.of(new ProducerRecord<>(TOPIC_1, MY_KEY, "1"), new ProducerRecord<>(TOPIC_1, MY_KEY, "2"), inCaseOfFailure()), Optional.empty());
-            consumeAndVerify(tester, this.clientConfig, TENANT_1_CLUSTER, TOPIC_1, groupId, new LinkedList<>(List.of(matchesRecord(TOPIC_1, MY_KEY, "1"))), true);
-            consumeAndVerify(tester, this.clientConfig, TENANT_1_CLUSTER, TOPIC_1, groupId, new LinkedList<>(List.of(matchesRecord(TOPIC_1, MY_KEY, "2"))), true);
+            final String topicName = tester.createTopic(TENANT_1_CLUSTER);
+            produceAndAssert(tester, this.clientConfig, TENANT_1_CLUSTER,
+                    Stream.of(new ProducerRecord<>(topicName, MY_KEY, "1"), new ProducerRecord<>(topicName, MY_KEY, "2"), inCaseOfFailure()), Optional.empty());
+            consumeAndAssert(tester, this.clientConfig, TENANT_1_CLUSTER, topicName, groupId, new LinkedList<>(List.of(matchesRecord(topicName, MY_KEY, "1"))), true);
+            consumeAndAssert(tester, this.clientConfig, TENANT_1_CLUSTER, topicName, groupId, new LinkedList<>(List.of(matchesRecord(topicName, MY_KEY, "2"))), true);
         }
     }
 
     @Test
     void alterOffsetCommit(KafkaCluster cluster) throws Exception {
         var config = getConfig(cluster, this.certificateGenerator);
-        try (var tester = kroxyliciousTester(config);
+        try (var tester = buildTester(config);
                 var admin = tester.admin(TENANT_1_CLUSTER, this.clientConfig)) {
             var groupId = testInfo.getDisplayName();
-            createTopics(admin, NEW_TOPIC_1);
-            produceAndVerify(tester, this.clientConfig, TENANT_1_CLUSTER,
-                    Stream.of(new ProducerRecord<>(TOPIC_1, MY_KEY, "1"), new ProducerRecord<>(TOPIC_1, MY_KEY, "2"), inCaseOfFailure()), Optional.empty());
-            consumeAndVerify(tester, this.clientConfig, TENANT_1_CLUSTER, TOPIC_1, groupId, new LinkedList<>(List.of(matchesRecord(TOPIC_1, MY_KEY, "1"))), true);
+            final String topicName = tester.createTopic(TENANT_1_CLUSTER);
+            produceAndAssert(tester, this.clientConfig, TENANT_1_CLUSTER,
+                    Stream.of(new ProducerRecord<>(topicName, MY_KEY, "1"), new ProducerRecord<>(topicName, MY_KEY, "2"), inCaseOfFailure()), Optional.empty());
+            consumeAndAssert(tester, this.clientConfig, TENANT_1_CLUSTER, topicName, groupId, new LinkedList<>(List.of(matchesRecord(topicName, MY_KEY, "1"))), true);
             var rememberedOffsets = admin.listConsumerGroupOffsets(groupId).partitionsToOffsetAndMetadata().get();
-            consumeAndVerify(tester, this.clientConfig, TENANT_1_CLUSTER, TOPIC_1, groupId, new LinkedList<>(List.of(matchesRecord(TOPIC_1, MY_KEY, "2"))), true);
+            consumeAndAssert(tester, this.clientConfig, TENANT_1_CLUSTER, topicName, groupId, new LinkedList<>(List.of(matchesRecord(topicName, MY_KEY, "2"))), true);
 
             admin.alterConsumerGroupOffsets(groupId, rememberedOffsets).all().get();
-            consumeAndVerify(tester, this.clientConfig, TENANT_1_CLUSTER, TOPIC_1, groupId, new LinkedList<>(List.of(matchesRecord(TOPIC_1, MY_KEY, "2"))), true);
+            consumeAndAssert(tester, this.clientConfig, TENANT_1_CLUSTER, topicName, groupId, new LinkedList<>(List.of(matchesRecord(topicName, MY_KEY, "2"))), true);
         }
     }
 
     @Test
     void deleteConsumerGroupOffsets(KafkaCluster cluster) throws Exception {
         var config = getConfig(cluster, this.certificateGenerator);
-        try (var tester = kroxyliciousTester(config);
+        try (var tester = buildTester(config);
                 var admin = tester.admin(TENANT_1_CLUSTER, this.clientConfig)) {
             var groupId = testInfo.getDisplayName();
-            createTopics(admin, NEW_TOPIC_1);
-            produceAndVerify(tester, this.clientConfig, TENANT_1_CLUSTER, Stream.of(new ProducerRecord<>(TOPIC_1, MY_KEY, "1"), inCaseOfFailure()), Optional.empty());
-            consumeAndVerify(tester, this.clientConfig, TENANT_1_CLUSTER, TOPIC_1, groupId, new LinkedList<>(List.of(matchesRecord(TOPIC_1, MY_KEY, "1"))), true);
+            final String topicName = tester.createTopic(TENANT_1_CLUSTER);
+            produceAndAssert(tester, this.clientConfig, TENANT_1_CLUSTER, Stream.of(new ProducerRecord<>(topicName, MY_KEY, "1"), inCaseOfFailure()), Optional.empty());
+            consumeAndAssert(tester, this.clientConfig, TENANT_1_CLUSTER, topicName, groupId, new LinkedList<>(List.of(matchesRecord(topicName, MY_KEY, "1"))), true);
 
-            admin.deleteConsumerGroupOffsets(groupId, Set.of(new TopicPartition(NEW_TOPIC_1.name(), 0))).all().get();
-            consumeAndVerify(tester, this.clientConfig, TENANT_1_CLUSTER, TOPIC_1, groupId, new LinkedList<>(List.of(matchesRecord(TOPIC_1, MY_KEY, "1"))), true);
+            admin.deleteConsumerGroupOffsets(groupId, Set.of(new TopicPartition(topicName, 0))).all().get();
+            consumeAndAssert(tester, this.clientConfig, TENANT_1_CLUSTER, topicName, groupId, new LinkedList<>(List.of(matchesRecord(topicName, MY_KEY, "1"))), true);
         }
     }
 
@@ -250,12 +257,12 @@ class MultiTenantIT extends BaseMultiTenantIT {
     }
 
     @Test
-    void produceInTransaction(KafkaCluster cluster) throws Exception {
+    void produceInTransaction(KafkaCluster cluster) {
         var config = getConfig(cluster, this.certificateGenerator);
         try (var tester = kroxyliciousTester(config);
                 var admin = tester.admin(TENANT_1_CLUSTER, this.clientConfig)) {
             createTopics(admin, NEW_TOPIC_1);
-            produceAndVerify(tester, this.clientConfig, TENANT_1_CLUSTER,
+            produceAndAssert(tester, this.clientConfig, TENANT_1_CLUSTER,
                     Stream.of(new ProducerRecord<>(TOPIC_1, MY_KEY, "1")),
                     Optional.of("12345"));
         }
@@ -271,7 +278,7 @@ class MultiTenantIT extends BaseMultiTenantIT {
             var outputTopic = "output";
             createTopics(admin, new NewTopic(inputTopic, 1, (short) 1),
                     new NewTopic(outputTopic, 1, (short) 1));
-            produceAndVerify(tester, this.clientConfig, TENANT_1_CLUSTER,
+            produceAndAssert(tester, this.clientConfig, TENANT_1_CLUSTER,
                     Stream.of(new ProducerRecord<>(inputTopic, MY_KEY, "1"),
                             new ProducerRecord<>(inputTopic, MY_KEY, "2"),
                             new ProducerRecord<>(inputTopic, MY_KEY, "3")),
@@ -307,7 +314,7 @@ class MultiTenantIT extends BaseMultiTenantIT {
             }
 
             // now verify that output contains the expected values.
-            consumeAndVerify(tester, this.clientConfig, TENANT_1_CLUSTER, outputTopic, groupId, new LinkedList<>(
+            consumeAndAssert(tester, this.clientConfig, TENANT_1_CLUSTER, outputTopic, groupId, new LinkedList<>(
                     List.of(matchesRecord(outputTopic, MY_KEY, "1"),
                             matchesRecord(outputTopic, MY_KEY, "2"),
                             matchesRecord(outputTopic, MY_KEY, "3"))),
@@ -323,7 +330,7 @@ class MultiTenantIT extends BaseMultiTenantIT {
             try (var admin = tester.admin(TENANT_1_CLUSTER, this.clientConfig)) {
                 createTopics(admin, NEW_TOPIC_1);
                 var transactionalId = "12345";
-                produceAndVerify(tester, this.clientConfig, TENANT_1_CLUSTER,
+                produceAndAssert(tester, this.clientConfig, TENANT_1_CLUSTER,
                         Stream.of(new ProducerRecord<>(TOPIC_1, MY_KEY, "1")),
                         Optional.of(transactionalId));
 
@@ -343,14 +350,14 @@ class MultiTenantIT extends BaseMultiTenantIT {
                 var adminTenant2 = tester.admin(TENANT_2_CLUSTER, this.clientConfig)) {
             createTopics(adminTenant1, NEW_TOPIC_1);
             var tenant1TransactionId = "12345";
-            produceAndVerify(tester, this.clientConfig, TENANT_1_CLUSTER,
+            produceAndAssert(tester, this.clientConfig, TENANT_1_CLUSTER,
                     Stream.of(new ProducerRecord<>(TOPIC_1, MY_KEY, "1")),
                     Optional.of(tenant1TransactionId));
             verifyTransactionsWithList(adminTenant1, Set.of(tenant1TransactionId));
 
             createTopics(adminTenant2, NEW_TOPIC_2);
             var tenant2TransactionId = "54321";
-            produceAndVerify(tester, this.clientConfig, TENANT_2_CLUSTER,
+            produceAndAssert(tester, this.clientConfig, TENANT_2_CLUSTER,
                     Stream.of(new ProducerRecord<>(TOPIC_2, MY_KEY, "1")),
                     Optional.of(tenant2TransactionId));
             verifyTransactionsWithList(adminTenant2, Set.of(tenant2TransactionId));
@@ -454,4 +461,4 @@ class MultiTenantIT extends BaseMultiTenantIT {
             });
         }
     }
-}
+}
\ No newline at end of file
diff --git a/kroxylicious-test-tools/pom.xml b/kroxylicious-test-tools/pom.xml
index d6e4e533..07fc7e5d 100644
--- a/kroxylicious-test-tools/pom.xml
+++ b/kroxylicious-test-tools/pom.xml
@@ -12,6 +12,7 @@
     <modelVersion>4.0.0</modelVersion>
 
     <packaging>jar</packaging>
+
     <parent>
         <groupId>io.kroxylicious</groupId>
         <artifactId>kroxylicious-parent</artifactId>
@@ -46,6 +47,10 @@
             <groupId>org.apache.kafka</groupId>
             <artifactId>kafka-clients</artifactId>
         </dependency>
+        <dependency>
+            <groupId>info.schnatterer.moby-names-generator</groupId>
+            <artifactId>moby-names-generator</artifactId>
+        </dependency>
         <dependency>
             <groupId>io.netty</groupId>
             <artifactId>netty-common</artifactId>
diff --git a/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/tester/DefaultKroxyliciousTester.java b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/tester/DefaultKroxyliciousTester.java
index 3f42efd2..85d67e17 100644
--- a/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/tester/DefaultKroxyliciousTester.java
+++ b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/tester/DefaultKroxyliciousTester.java
@@ -10,11 +10,24 @@ import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
 
+import org.apache.kafka.clients.CommonClientConfigs;
 import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.CreateTopicsResult;
+import org.apache.kafka.clients.admin.NewTopic;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.common.config.SslConfigs;
+import org.apache.kafka.common.security.auth.SecurityProtocol;
 import org.apache.kafka.common.serialization.Serde;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -22,29 +35,37 @@ import org.slf4j.LoggerFactory;
 import io.kroxylicious.proxy.KafkaProxy;
 import io.kroxylicious.proxy.config.Configuration;
 import io.kroxylicious.proxy.config.ConfigurationBuilder;
+import io.kroxylicious.proxy.config.VirtualCluster;
+import io.kroxylicious.proxy.config.tls.Tls;
 import io.kroxylicious.test.client.KafkaClient;
 
 import edu.umd.cs.findbugs.annotations.NonNull;
+import edu.umd.cs.findbugs.annotations.Nullable;
+import info.schnatterer.mobynamesgenerator.MobyNamesGenerator;
 
 public class DefaultKroxyliciousTester implements KroxyliciousTester {
     private AutoCloseable proxy;
+
+    @SuppressWarnings("OptionalUsedAsFieldOrParameterType")
+    private final Optional<KroxyliciousTesterBuilder.TrustStoreConfiguration> trustStoreConfiguration;
+
     private final Configuration kroxyliciousConfig;
 
     private final Map<String, KroxyliciousClients> clients;
+    private final Map<String, Set<String>> topicsPerVirtualCluster;
 
     private final ClientFactory clientFactory;
 
     private static final Logger LOGGER = LoggerFactory.getLogger(DefaultKroxyliciousTester.class);
 
-    DefaultKroxyliciousTester(ConfigurationBuilder configurationBuilder) {
-        this(configurationBuilder, DefaultKroxyliciousTester::spawnProxy, (clusterName, bootstrapServers) -> new KroxyliciousClients(bootstrapServers));
-    }
-
-    DefaultKroxyliciousTester(ConfigurationBuilder configuration, Function<Configuration, AutoCloseable> kroxyliciousFactory, ClientFactory clientFactory) {
-        kroxyliciousConfig = configuration.build();
-        proxy = kroxyliciousFactory.apply(kroxyliciousConfig);
-        clients = new HashMap<>();
+    DefaultKroxyliciousTester(ConfigurationBuilder configurationBuilder, Function<Configuration, AutoCloseable> kroxyliciousFactory, ClientFactory clientFactory,
+                              @Nullable KroxyliciousTesterBuilder.TrustStoreConfiguration trustStoreConfiguration) {
+        this.kroxyliciousConfig = configurationBuilder.build();
+        this.proxy = kroxyliciousFactory.apply(kroxyliciousConfig);
+        this.trustStoreConfiguration = Optional.ofNullable(trustStoreConfiguration);
+        this.clients = new ConcurrentHashMap<>();
         this.clientFactory = clientFactory;
+        topicsPerVirtualCluster = new ConcurrentHashMap<>();
     }
 
     private KroxyliciousClients clients() {
@@ -60,12 +81,31 @@ public class DefaultKroxyliciousTester implements KroxyliciousTester {
     }
 
     private KroxyliciousClients clients(String virtualCluster) {
-        return clients.computeIfAbsent(virtualCluster, this::buildKroxyliciousClients);
+        return clients.computeIfAbsent(virtualCluster,
+                k -> clientFactory.build(virtualCluster, buildDefaultClientConfiguration(virtualCluster)));
     }
 
     @NonNull
-    private KroxyliciousClients buildKroxyliciousClients(String clusterName) {
-        return this.clientFactory.build(clusterName, KroxyliciousConfigUtils.bootstrapServersFor(clusterName, kroxyliciousConfig));
+    private Map<String, Object> buildDefaultClientConfiguration(String virtualCluster) {
+        Map<String, Object> defaultClientConfig = new HashMap<>();
+        defaultClientConfig.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, KroxyliciousConfigUtils.bootstrapServersFor(virtualCluster, kroxyliciousConfig));
+        configureClientTls(virtualCluster, defaultClientConfig);
+        return defaultClientConfig;
+    }
+
+    private void configureClientTls(String virtualCluster, Map<String, Object> defaultClientConfig) {
+        final VirtualCluster definedCluster = kroxyliciousConfig.virtualClusters().get(virtualCluster);
+        if (definedCluster != null) {
+            final Optional<Tls> tls = definedCluster.tls();
+            if (tls.isPresent()) {
+                defaultClientConfig.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, SecurityProtocol.SSL.name);
+                if (trustStoreConfiguration.isPresent()) {
+                    final KroxyliciousTesterBuilder.TrustStoreConfiguration storeConfiguration = trustStoreConfiguration.get();
+                    defaultClientConfig.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, storeConfiguration.trustStoreLocation());
+                    defaultClientConfig.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, storeConfiguration.trustStorePassword());
+                }
+            }
+        }
     }
 
     @Override
@@ -164,7 +204,7 @@ public class DefaultKroxyliciousTester implements KroxyliciousTester {
             proxy = spawnProxy(kroxyliciousConfig);
         }
         catch (Exception e) {
-            throw new RuntimeException(e);
+            throw new IllegalStateException(e);
         }
     }
 
@@ -173,12 +213,7 @@ public class DefaultKroxyliciousTester implements KroxyliciousTester {
         try {
             List<Exception> exceptions = new ArrayList<>();
             for (KroxyliciousClients c : clients.values()) {
-                try {
-                    c.close();
-                }
-                catch (Exception e) {
-                    exceptions.add(e);
-                }
+                closeClient(c).ifPresent(exceptions::add);
             }
             proxy.close();
             if (!exceptions.isEmpty()) {
@@ -188,7 +223,17 @@ public class DefaultKroxyliciousTester implements KroxyliciousTester {
             }
         }
         catch (Exception e) {
-            throw new RuntimeException(e);
+            throw new IllegalStateException(e);
+        }
+    }
+
+    private static Optional<Exception> closeClient(KroxyliciousClients c) {
+        try {
+            c.close();
+            return Optional.empty();
+        }
+        catch (Exception e) {
+            return Optional.of(e);
         }
     }
 
@@ -203,9 +248,62 @@ public class DefaultKroxyliciousTester implements KroxyliciousTester {
         return kafkaProxy;
     }
 
+    @Override
+    public Set<String> createTopics(String virtualCluster, int numberOfTopics) {
+        try (Admin admin = clients(virtualCluster).admin()) {
+            final List<NewTopic> newTopics = IntStream.range(0, numberOfTopics).mapToObj(ignored -> {
+                final String topicName = MobyNamesGenerator.getRandomName();
+                return new NewTopic(topicName, (short) 1, (short) 1);
+            }).toList();
+            final CreateTopicsResult createTopicsResult = admin.createTopics(newTopics);
+            createTopicsResult.all().get(30, TimeUnit.SECONDS);
+
+            // TODO should this be driven by result of the createTopic call?
+            final Set<String> topicNames = newTopics.stream().map(NewTopic::name).collect(Collectors.toSet());
+            topicsForVirtualCluster(virtualCluster).addAll(topicNames);
+            return topicNames;
+        }
+        catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            throw new IllegalStateException("Failed to create topics on " + virtualCluster, e);
+        }
+        catch (TimeoutException e) {
+            throw new IllegalStateException("Timed out creating topics on " + virtualCluster, e);
+        }
+        catch (ExecutionException e) {
+            throw new IllegalStateException("Failed to create topics on " + virtualCluster, e.getCause());
+        }
+    }
+
+    @Override
+    public void deleteTopics(String virtualCluster) {
+        try (Admin admin = clients(virtualCluster).admin()) {
+            final Set<String> topics = topicsForVirtualCluster(virtualCluster);
+            if (!topics.isEmpty()) {
+                admin.deleteTopics(topics)
+                        .all()
+                        .get(30, TimeUnit.SECONDS);
+            }
+        }
+        catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            throw new IllegalStateException("Failed to delete topics on " + virtualCluster, e);
+        }
+        catch (ExecutionException e) {
+            throw new IllegalStateException("Failed to delete topics on " + virtualCluster, e.getCause());
+        }
+        catch (TimeoutException e) {
+            throw new IllegalStateException("Timed out deleting topics on " + virtualCluster, e);
+        }
+    }
+
+    private Set<String> topicsForVirtualCluster(String clusterName) {
+        return topicsPerVirtualCluster.computeIfAbsent(clusterName, key -> ConcurrentHashMap.newKeySet());
+    }
+
     @FunctionalInterface
     interface ClientFactory {
-        KroxyliciousClients build(String clusterName, String bootstrapServers);
+        KroxyliciousClients build(String clusterName, Map<String, Object> defaultClientConfig);
     }
 
 }
diff --git a/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/tester/KroxyliciousClients.java b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/tester/KroxyliciousClients.java
index 32e4211e..ecac29ea 100644
--- a/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/tester/KroxyliciousClients.java
+++ b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/tester/KroxyliciousClients.java
@@ -31,20 +31,20 @@ import io.kroxylicious.testing.kafka.clients.CloseableConsumer;
 import io.kroxylicious.testing.kafka.clients.CloseableProducer;
 
 class KroxyliciousClients {
-    private final String bootstrapServers;
+    private final Map<String, Object> defaultClientConfiguration;
 
     private final List<Admin> admins;
     private final List<Producer<?, ?>> producers;
     private final List<Consumer<?, ?>> consumers;
     private final ClientFactory clientFactory;
 
-    KroxyliciousClients(String bootstrapServers) {
-        this(bootstrapServers, new ClientFactory() {
+    KroxyliciousClients(Map<String, Object> defaultClientConfiguration) {
+        this(defaultClientConfiguration, new ClientFactory() {
         });
     }
 
-    KroxyliciousClients(String bootstrapServers, ClientFactory clientFactory) {
-        this.bootstrapServers = bootstrapServers;
+    KroxyliciousClients(Map<String, Object> defaultClientConfiguration, ClientFactory clientFactory) {
+        this.defaultClientConfiguration = defaultClientConfiguration;
         this.admins = new ArrayList<>();
         this.producers = new ArrayList<>();
         this.consumers = new ArrayList<>();
@@ -52,7 +52,7 @@ class KroxyliciousClients {
     }
 
     public Admin admin(Map<String, Object> additionalConfig) {
-        Map<String, Object> config = createConfigMap(bootstrapServers, additionalConfig);
+        Map<String, Object> config = createClientConfig(additionalConfig);
         Admin admin = clientFactory.newAdmin(config);
         admins.add(admin);
         return admin;
@@ -71,7 +71,7 @@ class KroxyliciousClients {
     }
 
     public <U, V> Producer<U, V> producer(Serde<U> keySerde, Serde<V> valueSerde, Map<String, Object> additionalConfig) {
-        Map<String, Object> config = createConfigMap(bootstrapServers, additionalConfig);
+        Map<String, Object> config = createClientConfig(additionalConfig);
         Producer<U, V> producer = this.clientFactory.newProducer(config, keySerde.serializer(), valueSerde.serializer());
         producers.add(producer);
         return producer;
@@ -86,14 +86,14 @@ class KroxyliciousClients {
     }
 
     public <U, V> Consumer<U, V> consumer(Serde<U> keySerde, Serde<V> valueSerde, Map<String, Object> additionalConfig) {
-        Map<String, Object> config = createConfigMap(bootstrapServers, additionalConfig);
+        Map<String, Object> config = createClientConfig(additionalConfig);
         Consumer<U, V> consumer = clientFactory.newConsumer(config, keySerde.deserializer(), valueSerde.deserializer());
         consumers.add(consumer);
         return consumer;
     }
 
     public KafkaClient simpleTestClient() {
-        String[] hostPort = bootstrapServers.split(":");
+        String[] hostPort = defaultClientConfiguration.get(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG).toString().split(":");
         return new KafkaClient(hostPort[0], Integer.parseInt(hostPort[1]));
     }
 
@@ -126,9 +126,8 @@ class KroxyliciousClients {
         return exceptions;
     }
 
-    private Map<String, Object> createConfigMap(String bootstrapServers, Map<String, Object> additionalConfig) {
-        Map<String, Object> config = new HashMap<>();
-        config.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
+    private Map<String, Object> createClientConfig(Map<String, Object> additionalConfig) {
+        Map<String, Object> config = new HashMap<>(defaultClientConfiguration);
         config.putAll(additionalConfig);
         return config;
     }
diff --git a/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/tester/KroxyliciousConfigUtils.java b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/tester/KroxyliciousConfigUtils.java
index 7e7b1a5d..a7af26ad 100644
--- a/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/tester/KroxyliciousConfigUtils.java
+++ b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/tester/KroxyliciousConfigUtils.java
@@ -26,7 +26,7 @@ public class KroxyliciousConfigUtils {
     }
 
     public static final String DEFAULT_VIRTUAL_CLUSTER = "demo";
-    private static final HostPort DEFAULT_PROXY_BOOTSTRAP = new HostPort("localhost", 9192);
+    static final HostPort DEFAULT_PROXY_BOOTSTRAP = new HostPort("localhost", 9192);
 
     /**
      * Create a KroxyliciousConfigBuilder with a single virtual cluster configured to
diff --git a/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/tester/KroxyliciousTester.java b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/tester/KroxyliciousTester.java
index 0c1c8b27..8f804529 100644
--- a/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/tester/KroxyliciousTester.java
+++ b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/tester/KroxyliciousTester.java
@@ -8,6 +8,7 @@ package io.kroxylicious.test.tester;
 
 import java.io.Closeable;
 import java.util.Map;
+import java.util.Set;
 
 import org.apache.kafka.clients.admin.Admin;
 import org.apache.kafka.clients.consumer.Consumer;
@@ -219,4 +220,36 @@ public interface KroxyliciousTester extends Closeable {
      */
     void close();
 
-}
+    /**
+     * Creates a single topic on the default Kafka cluster with a fixed partition count (1) and replication factor (1).
+     * <p>
+     * The number of partitions can be increased via {@link  Admin#createPartitions(Map) Admin.createParitions}.
+     * The number of replicas can be increased via {@link Admin#alterPartitionReassignments(Map) Admin.alterParitionsReassignments} by altering the replica assignments.
+     * See the <a href="https://kafka.apache.org/documentation/#basic_ops_increase_replication_factor"> Kafka docs</a> for details
+     * <p>
+     * @param clusterName the name of the virtual cluster on which to create the topic
+     * @return the name of the created topic
+     */
+    default String createTopic(String clusterName) {
+        return createTopics(clusterName, 1).stream().findFirst().orElseThrow(() -> new IllegalStateException("Failed to create topic"));
+    }
+
+    /**
+     * Creates N topics with a fixed partition count (1) and replication factor (1).
+     * <p>
+     * The number of partitions can be increased via {@link  Admin#createPartitions(Map) Admin.createParitions}.
+     * The number of replicas can be increased via {@link Admin#alterPartitionReassignments(Map) Admin.alterParitionsReassignments} by altering the replica assignments.
+     * See the <a href="https://kafka.apache.org/documentation/#basic_ops_increase_replication_factor"> Kafka docs</a> for details
+     * <p>
+     * @param clusterName the name of the virtual cluster on which to create the topic
+     * @param numberOfTopics the number of topics to create on the cluster
+     * @return the Set of topic names which have been created.
+     */
+    Set<String> createTopics(String clusterName, int numberOfTopics);
+
+    /**
+     * Asks the tester to delete all topics <strong>created by</strong> the tester on a specific virtual cluster.
+     * @param clusterName the name of the virtual cluster from which to delete the topics.
+     */
+    void deleteTopics(String clusterName);
+}
\ No newline at end of file
diff --git a/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/tester/KroxyliciousTesterBuilder.java b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/tester/KroxyliciousTesterBuilder.java
new file mode 100644
index 00000000..97e78f5d
--- /dev/null
+++ b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/tester/KroxyliciousTesterBuilder.java
@@ -0,0 +1,100 @@
+/*
+ * 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.test.tester;
+
+import java.util.function.Function;
+
+import io.kroxylicious.proxy.config.Configuration;
+import io.kroxylicious.proxy.config.ConfigurationBuilder;
+
+import edu.umd.cs.findbugs.annotations.Nullable;
+
+/**
+ * Kroxylicious strives to offer high quality testing tools which allow filter authors to focus on their business logic.
+ * However, as Kroxylicious is very configurable so is the Tester. Most of the time users do not care about a lot of the configuration options
+ * and thus a builder, which can provide sensible defaults, provides the simplest API to configure things.
+ */
+public class KroxyliciousTesterBuilder {
+
+    private String trustStoreLocation = null;
+    @Nullable
+    private String trustStorePassword = null;
+    private Function<Configuration, AutoCloseable> kroxyliciousFactory = DefaultKroxyliciousTester::spawnProxy;
+    private DefaultKroxyliciousTester.ClientFactory clientFactory = (clusterName, defaultConfiguration) -> new KroxyliciousClients(defaultConfiguration);
+    private ConfigurationBuilder configurationBuilder;
+
+    /**
+     * Supplies the builder used to configure the kroxylicious instance to be used in tests.
+     * @param configurationBuilder the builder instance ussd to configure the kroxylicious instance.
+     * @return the current instance of the builder for chaining in a fluent fashion.
+     */
+    public KroxyliciousTesterBuilder setConfigurationBuilder(ConfigurationBuilder configurationBuilder) {
+        this.configurationBuilder = configurationBuilder;
+        return this;
+    }
+
+    /**
+     * Configure the location of the trust store.
+     * <p>
+     * <strong>Required when using SSL/TLS</strong>
+     * @param trustStoreLocation the path to the trust store to be used when clients connect to an SSL enabled Kroxylicious instance.
+     * @return the current instance of the builder for chaining in a fluent fashion.
+     */
+    public KroxyliciousTesterBuilder setTrustStoreLocation(String trustStoreLocation) {
+        this.trustStoreLocation = trustStoreLocation;
+        return this;
+    }
+
+    /**
+     * Configure the password of the trust store.
+     * <p>
+     * <strong>Optional when using SSL/TLS</strong>
+     * @param trustStorePassword the password used to secure the trust store.
+     * @return the current instance of the builder for chaining in a fluent fashion.
+     */
+    public KroxyliciousTesterBuilder setTrustStorePassword(@Nullable String trustStorePassword) {
+        this.trustStorePassword = trustStorePassword;
+        return this;
+    }
+
+    /**
+     * Explicitly creates a specific type of tester.
+     * @return A {@link DefaultKroxyliciousTester} configured with the options specified by the builder.
+     */
+    public DefaultKroxyliciousTester createDefaultKroxyliciousTester() {
+        final TrustStoreConfiguration trustStoreConfiguration = trustStoreLocation != null ? new TrustStoreConfiguration(trustStoreLocation, trustStorePassword) : null;
+        return new DefaultKroxyliciousTester(configurationBuilder, kroxyliciousFactory, clientFactory, trustStoreConfiguration);
+    }
+
+    /**
+     * Supplies a function which translates Kroxylicious config into a kroxylicious instance.
+     * <p>
+     * <em>Intended for internal use to allow unit testing of the testers.</em>
+     *
+     * @param kroxyliciousFactory config translation function.
+     * @return the current instance of the builder for chaining in a fluent fashion.
+     */
+    KroxyliciousTesterBuilder setKroxyliciousFactory(Function<Configuration, AutoCloseable> kroxyliciousFactory) {
+        this.kroxyliciousFactory = kroxyliciousFactory;
+        return this;
+    }
+
+    /**
+     * Supplies a custom factory for creating Kafka Clients.
+     * <p>
+     * <em>Intended for internal use to allow unit testing of the testers.</em>
+     *
+     * @param clientFactory a factory for creating Kafka Client instances.
+     * @return the current instance of the builder for chaining in a fluent fashion.
+     */
+    KroxyliciousTesterBuilder setClientFactory(DefaultKroxyliciousTester.ClientFactory clientFactory) {
+        this.clientFactory = clientFactory;
+        return this;
+    }
+
+    record TrustStoreConfiguration(String trustStoreLocation, @Nullable String trustStorePassword) {}
+}
\ No newline at end of file
diff --git a/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/tester/KroxyliciousTesters.java b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/tester/KroxyliciousTesters.java
index 81ea4cf4..ec84422a 100644
--- a/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/tester/KroxyliciousTesters.java
+++ b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/tester/KroxyliciousTesters.java
@@ -25,7 +25,18 @@ public class KroxyliciousTesters {
      * @return KroxyliciousTester
      */
     public static KroxyliciousTester kroxyliciousTester(ConfigurationBuilder builder) {
-        return new DefaultKroxyliciousTester(builder);
+        return new KroxyliciousTesterBuilder().setConfigurationBuilder(builder).createDefaultKroxyliciousTester();
+    }
+
+    /**
+     * Creates a builder of kroxylicious testers with given KroxyliciousConfigBuilder.
+     * This will create and start an in-process kroxylicious instance, it is
+     * up to the client to close it.
+     * @param builder configuration builder for the kroxylicious instance
+     * @return KroxyliciousTester
+     */
+    public static KroxyliciousTesterBuilder newBuilder(ConfigurationBuilder builder) {
+        return new KroxyliciousTesterBuilder().setConfigurationBuilder(builder);
     }
 
     /**
@@ -33,12 +44,13 @@ public class KroxyliciousTesters {
      * a function to build an AutoClosable. This is to enable clients to provide their
      * own custom code to create and start a Kroxylicious instance (for example to
      * create it as a sub-process instead of in-process).
-     * @param builder configuration builder for the kroxylicious instance
+     * @param configurationBuilder configuration builder for the kroxylicious instance
      * @param kroxyliciousFactory factory that takes a KroxyliciousConfig and is responsible for starting a Kroxylicious instance for that config
      * @return KroxyliciousTester
      */
-    public static KroxyliciousTester kroxyliciousTester(ConfigurationBuilder builder, Function<Configuration, AutoCloseable> kroxyliciousFactory) {
-        return new DefaultKroxyliciousTester(builder, kroxyliciousFactory, (clusterName, bootstrapServers) -> new KroxyliciousClients(bootstrapServers));
+    public static KroxyliciousTester kroxyliciousTester(ConfigurationBuilder configurationBuilder, Function<Configuration, AutoCloseable> kroxyliciousFactory) {
+        return new KroxyliciousTesterBuilder().setConfigurationBuilder(configurationBuilder).setKroxyliciousFactory(kroxyliciousFactory)
+                .setClientFactory((clusterName, defaultClientConfiguration) -> new KroxyliciousClients(defaultClientConfiguration)).createDefaultKroxyliciousTester();
     }
 
     /**
@@ -49,6 +61,8 @@ public class KroxyliciousTesters {
      * @return KroxyliciousTester
      */
     public static MockServerKroxyliciousTester mockKafkaKroxyliciousTester(Function<String, ConfigurationBuilder> configurationForMockBootstrap) {
-        return new MockServerKroxyliciousTester(MockServer.startOnRandomPort(), configurationForMockBootstrap);
+        return new MockServerKroxyliciousTester(MockServer.startOnRandomPort(), configurationForMockBootstrap, DefaultKroxyliciousTester::spawnProxy,
+                (clusterName, defaultClientConfiguration) -> new KroxyliciousClients(defaultClientConfiguration), null);
     }
+
 }
\ No newline at end of file
diff --git a/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/tester/MockServerKroxyliciousTester.java b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/tester/MockServerKroxyliciousTester.java
index 7302aa14..c6b55103 100644
--- a/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/tester/MockServerKroxyliciousTester.java
+++ b/kroxylicious-test-tools/src/main/java/io/kroxylicious/test/tester/MockServerKroxyliciousTester.java
@@ -15,11 +15,14 @@ import org.hamcrest.Description;
 import org.hamcrest.Matcher;
 import org.hamcrest.TypeSafeMatcher;
 
+import io.kroxylicious.proxy.config.Configuration;
 import io.kroxylicious.proxy.config.ConfigurationBuilder;
 import io.kroxylicious.test.Request;
 import io.kroxylicious.test.ResponsePayload;
 import io.kroxylicious.test.server.MockServer;
 
+import edu.umd.cs.findbugs.annotations.Nullable;
+
 import static org.apache.kafka.common.protocol.ApiKeys.PRODUCE;
 
 /**
@@ -33,8 +36,10 @@ public class MockServerKroxyliciousTester extends DefaultKroxyliciousTester {
 
     private final MockServer mockServer;
 
-    MockServerKroxyliciousTester(MockServer mockServer, Function<String, ConfigurationBuilder> configurationForMockBootstrap) {
-        super(configurationForMockBootstrap.apply("localhost:" + mockServer.port()));
+    MockServerKroxyliciousTester(MockServer mockServer, Function<String, ConfigurationBuilder> configurationForMockBootstrap,
+                                 Function<Configuration, AutoCloseable> kroxyliciousFactory, ClientFactory clientFactory,
+                                 @Nullable KroxyliciousTesterBuilder.TrustStoreConfiguration trustStoreConfiguration) {
+        super(configurationForMockBootstrap.apply("localhost:" + mockServer.port()), kroxyliciousFactory, clientFactory, trustStoreConfiguration);
         this.mockServer = mockServer;
     }
 
diff --git a/kroxylicious-test-tools/src/test/java/io/kroxylicious/test/tester/DefaultKroxyliciousTesterTest.java b/kroxylicious-test-tools/src/test/java/io/kroxylicious/test/tester/DefaultKroxyliciousTesterTest.java
index d4346927..ce1dc6d5 100644
--- a/kroxylicious-test-tools/src/test/java/io/kroxylicious/test/tester/DefaultKroxyliciousTesterTest.java
+++ b/kroxylicious-test-tools/src/test/java/io/kroxylicious/test/tester/DefaultKroxyliciousTesterTest.java
@@ -6,26 +6,60 @@
 
 package io.kroxylicious.test.tester;
 
+import java.io.IOException;
+import java.nio.file.Path;
+import java.security.GeneralSecurityException;
+import java.util.Collection;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.function.Consumer;
+
+import org.apache.kafka.clients.CommonClientConfigs;
 import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.CreateTopicsResult;
+import org.apache.kafka.clients.admin.DeleteTopicsResult;
 import org.apache.kafka.clients.consumer.MockConsumer;
 import org.apache.kafka.clients.consumer.OffsetResetStrategy;
 import org.apache.kafka.clients.producer.MockProducer;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.KafkaFuture;
+import org.apache.kafka.common.config.SslConfigs;
+import org.apache.kafka.common.security.auth.SecurityProtocol;
+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.junit.jupiter.api.io.TempDir;
 import org.mockito.Mock;
+import org.mockito.hamcrest.MockitoHamcrest;
 import org.mockito.junit.jupiter.MockitoExtension;
 
+import io.kroxylicious.proxy.config.ClusterNetworkAddressConfigProviderDefinitionBuilder;
+import io.kroxylicious.proxy.config.ConfigurationBuilder;
+import io.kroxylicious.proxy.config.VirtualClusterBuilder;
+import io.kroxylicious.proxy.internal.clusternetworkaddressconfigprovider.PortPerBrokerClusterNetworkAddressConfigProvider;
+import io.kroxylicious.testing.kafka.common.KeytoolCertificateGenerator;
+
 import edu.umd.cs.findbugs.annotations.NonNull;
 
+import static io.kroxylicious.test.tester.KroxyliciousConfigUtils.DEFAULT_PROXY_BOOTSTRAP;
 import static io.kroxylicious.test.tester.KroxyliciousConfigUtils.DEFAULT_VIRTUAL_CLUSTER;
 import static io.kroxylicious.test.tester.KroxyliciousConfigUtils.proxy;
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.junit.jupiter.api.Assertions.fail;
-import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyCollection;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyMap;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mock.Strictness.LENIENT;
 import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
@@ -37,6 +71,8 @@ class DefaultKroxyliciousTesterTest {
     private static final String VIRTUAL_CLUSTER_C = "clusterC";
     private static final String EXCEPTION_MESSAGE = "KaBOOM!!";
     private static final String DEFAULT_CLUSTER = "demo";
+    private static final String TLS_CLUSTER = "secureCluster";
+    public static final int TOPIC_COUNT = 5;
     String backingCluster = "broker01.example.com:9090";
 
     @Mock(strictness = LENIENT)
@@ -63,10 +99,11 @@ class DefaultKroxyliciousTesterTest {
 
     @BeforeEach
     void setUp() {
-        when(clientFactory.build(eq(DEFAULT_VIRTUAL_CLUSTER), anyString())).thenReturn(kroxyliciousClients);
-        when(clientFactory.build(eq(VIRTUAL_CLUSTER_A), anyString())).thenReturn(kroxyliciousClientsA);
-        when(clientFactory.build(eq(VIRTUAL_CLUSTER_B), anyString())).thenReturn(kroxyliciousClientsB);
-        when(clientFactory.build(eq(VIRTUAL_CLUSTER_C), anyString())).thenReturn(kroxyliciousClientsC);
+        when(clientFactory.build(eq(DEFAULT_VIRTUAL_CLUSTER), anyMap())).thenReturn(kroxyliciousClients);
+        when(clientFactory.build(eq(TLS_CLUSTER), anyMap())).thenReturn(kroxyliciousClients);
+        when(clientFactory.build(eq(VIRTUAL_CLUSTER_A), anyMap())).thenReturn(kroxyliciousClientsA);
+        when(clientFactory.build(eq(VIRTUAL_CLUSTER_B), anyMap())).thenReturn(kroxyliciousClientsB);
+        when(clientFactory.build(eq(VIRTUAL_CLUSTER_C), anyMap())).thenReturn(kroxyliciousClientsC);
         when(kroxyliciousClients.admin()).thenReturn(admin);
         when(kroxyliciousClients.producer()).thenReturn(producer);
         when(kroxyliciousClients.consumer()).thenReturn(consumer);
@@ -83,7 +120,7 @@ class DefaultKroxyliciousTesterTest {
 
             // Then
             // In theory the bootstrap address is predicable but asserting it is not part of this test
-            verify(clientFactory).build(eq(DEFAULT_CLUSTER), anyString());
+            verify(clientFactory).build(eq(DEFAULT_CLUSTER), anyMap());
             verify(kroxyliciousClients).admin();
         }
     }
@@ -99,7 +136,7 @@ class DefaultKroxyliciousTesterTest {
 
             // Then
             // In theory the bootstrap address is predicable but asserting it is not part of this test
-            verify(clientFactory).build(eq(DEFAULT_CLUSTER), anyString());
+            verify(clientFactory).build(eq(DEFAULT_CLUSTER), anyMap());
             verify(kroxyliciousClients).admin();
         }
     }
@@ -115,7 +152,7 @@ class DefaultKroxyliciousTesterTest {
 
             // Then
             // In theory the bootstrap address is predicable but asserting it is not part of this test
-            verify(clientFactory).build(eq(DEFAULT_CLUSTER), anyString());
+            verify(clientFactory).build(eq(DEFAULT_CLUSTER), anyMap());
             verify(kroxyliciousClients).producer();
         }
     }
@@ -131,7 +168,7 @@ class DefaultKroxyliciousTesterTest {
 
             // Then
             // In theory the bootstrap address is predicable but asserting it is not part of this test
-            verify(clientFactory).build(eq(DEFAULT_CLUSTER), anyString());
+            verify(clientFactory).build(eq(DEFAULT_CLUSTER), anyMap());
             verify(kroxyliciousClients).producer();
         }
     }
@@ -147,7 +184,7 @@ class DefaultKroxyliciousTesterTest {
 
             // Then
             // In theory the bootstrap address is predicable but asserting it is not part of this test
-            verify(clientFactory).build(eq(DEFAULT_CLUSTER), anyString());
+            verify(clientFactory).build(eq(DEFAULT_CLUSTER), anyMap());
             verify(kroxyliciousClients).consumer();
         }
     }
@@ -163,7 +200,7 @@ class DefaultKroxyliciousTesterTest {
 
             // Then
             // In theory the bootstrap address is predicable but asserting it is not part of this test
-            verify(clientFactory).build(eq(DEFAULT_CLUSTER), anyString());
+            verify(clientFactory).build(eq(DEFAULT_CLUSTER), anyMap());
             verify(kroxyliciousClients).consumer();
         }
     }
@@ -178,7 +215,6 @@ class DefaultKroxyliciousTesterTest {
             tester.close();
 
             // Then
-            // In theory the bootstrap address is predicable but asserting it is not part of this test
             verify(kroxyliciousClients).close();
         }
     }
@@ -226,16 +262,303 @@ class DefaultKroxyliciousTesterTest {
         }
     }
 
+    @Test
+    void shouldConfigureConsumerForTls(@TempDir Path certsDirectory) throws IOException {
+        // Given
+        final String certFilePath = certsDirectory.resolve(Path.of("cert-file")).toAbsolutePath().toString();
+        final String trustStorePath = certsDirectory.resolve(Path.of("trust-store")).toAbsolutePath().toString();
+        var keytoolCertificateGenerator = new KeytoolCertificateGenerator(certFilePath, trustStorePath);
+        try (var tester = buildSecureTester(keytoolCertificateGenerator)) {
+
+            // When
+            tester.consumer(TLS_CLUSTER);
+            tester.producer(TLS_CLUSTER);
+            tester.admin(TLS_CLUSTER);
+
+            // Then
+
+            verify(clientFactory).build(eq(TLS_CLUSTER), argThat(assertSslConfiguration(trustStorePath)));
+        }
+    }
+
+    @Test
+    void shouldConfigureProducerForTls(@TempDir Path certsDirectory) throws IOException {
+        // Given
+        final String certFilePath = certsDirectory.resolve(Path.of("cert-file")).toAbsolutePath().toString();
+        final String trustStorePath = certsDirectory.resolve(Path.of("trust-store")).toAbsolutePath().toString();
+        var keytoolCertificateGenerator = new KeytoolCertificateGenerator(certFilePath, trustStorePath);
+        try (var tester = buildSecureTester(keytoolCertificateGenerator)) {
+
+            // When
+            tester.producer(TLS_CLUSTER);
+
+            // Then
+            verify(clientFactory).build(eq(TLS_CLUSTER), argThat(assertSslConfiguration(trustStorePath)));
+        }
+    }
+
+    @Test
+    void shouldConfigureAdminForTls(@TempDir Path certsDirectory) throws IOException {
+        // Given
+        final String certFilePath = certsDirectory.resolve(Path.of("cert-file")).toAbsolutePath().toString();
+        final String trustStorePath = certsDirectory.resolve(Path.of("trust-store")).toAbsolutePath().toString();
+        var keytoolCertificateGenerator = new KeytoolCertificateGenerator(certFilePath, trustStorePath);
+        try (var tester = buildSecureTester(keytoolCertificateGenerator)) {
+
+            // When
+            tester.admin(TLS_CLUSTER);
+
+            // Then
+            verify(clientFactory).build(eq(TLS_CLUSTER), argThat(assertSslConfiguration(trustStorePath)));
+        }
+    }
+
+    @Test
+    void shouldCreateSingleTopic() {
+        // Given
+        final CreateTopicsResult createTopicsResult = mock(CreateTopicsResult.class);
+        when(admin.createTopics(anyCollection())).thenReturn(createTopicsResult);
+        when(createTopicsResult.all()).thenReturn(KafkaFuture.completedFuture(null));
+        try (KroxyliciousTester tester = buildDefaultTester()) {
+
+            // When
+            final String actualTopicName = tester.createTopic(DEFAULT_CLUSTER);
+
+            // Then
+            verify(admin).createTopics(argThat(topics -> assertThat(topics).hasSize(1)));
+            assertThat(actualTopicName).isNotBlank();
+        }
+    }
+
+    @Test
+    void shouldCreateMultipleTopics() {
+        // Given
+        final CreateTopicsResult createTopicsResult = mock(CreateTopicsResult.class);
+        when(admin.createTopics(anyCollection())).thenReturn(createTopicsResult);
+        when(createTopicsResult.all()).thenReturn(KafkaFuture.completedFuture(null));
+        try (KroxyliciousTester tester = buildDefaultTester()) {
+
+            // When
+            final Set<String> actualTopicName = tester.createTopics(DEFAULT_CLUSTER, TOPIC_COUNT);
+
+            // Then
+            verify(admin).createTopics(argThat(topics -> assertThat(topics).hasSize(TOPIC_COUNT)));
+            assertThat(actualTopicName).isNotEmpty().hasSize(TOPIC_COUNT);
+        }
+    }
+
+    @Test
+    void shouldTrackTopicsForCluster() {
+        // Given
+        final Admin adminB = mock(Admin.class);
+        allowCreateTopic(kroxyliciousClientsA, admin);
+        allowCreateTopic(kroxyliciousClientsB, adminB);
+
+        try (KroxyliciousTester tester = buildTester()) {
+            tester.createTopics(VIRTUAL_CLUSTER_A, TOPIC_COUNT);
+
+            // When
+            tester.createTopics(VIRTUAL_CLUSTER_B, TOPIC_COUNT);
+
+            // Then
+            verify(admin).createTopics(argThat(topics -> assertThat(topics).hasSize(TOPIC_COUNT)));
+            verify(adminB).createTopics(argThat(topics -> assertThat(topics).hasSize(TOPIC_COUNT)));
+        }
+    }
+
+    @Test
+    void shouldDeleteTopicsForCluster() {
+        // Given
+        allowCreateTopic(kroxyliciousClientsA, admin);
+        final DeleteTopicsResult deleteTopicsResult = mock(DeleteTopicsResult.class);
+        when(admin.deleteTopics(anyCollection())).thenReturn(deleteTopicsResult);
+        when(deleteTopicsResult.all()).thenReturn(KafkaFuture.completedFuture(null));
+
+        try (KroxyliciousTester tester = buildTester()) {
+            tester.createTopics(VIRTUAL_CLUSTER_A, TOPIC_COUNT);
+
+            // When
+            tester.deleteTopics(VIRTUAL_CLUSTER_A);
+
+            // Then
+            verify(admin).deleteTopics(DefaultKroxyliciousTesterTest.<Collection<String>> argThat(topics -> assertThat(topics).hasSize(TOPIC_COUNT)));
+        }
+    }
+
+    @Test
+    void shouldNotThrowWhenNoTopicsCreated() {
+        // Given
+        when(kroxyliciousClientsA.admin()).thenReturn(admin);
+
+        try (KroxyliciousTester tester = buildTester()) {
+            // When
+            tester.deleteTopics(VIRTUAL_CLUSTER_A);
+
+            // Then
+            //We can't use verifyNoInteractions or verifyNoMoreInteractions here as the try-with-resource block will trigger admin.close()
+            verify(admin, times(0)).deleteTopics(anyCollection());
+        }
+    }
+
+    @SuppressWarnings("unchecked")
+    @Test
+    void shouldPropagateErrorsCreatingTopics() throws ExecutionException, InterruptedException, TimeoutException {
+        // Given
+        final CreateTopicsResult createTopicsResult = mock(CreateTopicsResult.class);
+        final KafkaFuture<Void> kafkaFuture = mock(KafkaFuture.class);
+        when(admin.createTopics(anyCollection())).thenReturn(createTopicsResult);
+        when(createTopicsResult.all()).thenReturn(kafkaFuture);
+        when(kafkaFuture.get(anyLong(), any(TimeUnit.class))).thenThrow(new ExecutionException("Kafka is nae working pal", new KafkaException("KaBoom")));
+
+        try (KroxyliciousTester tester = buildDefaultTester()) {
+
+            // When
+            assertThatThrownBy(() -> tester.createTopic(DEFAULT_CLUSTER))
+                    // Then
+                    .isInstanceOf(IllegalStateException.class)
+                    .hasMessage("Failed to create topics on " + DEFAULT_CLUSTER)
+                    .hasCauseInstanceOf(KafkaException.class);
+        }
+    }
+
+    @SuppressWarnings("unchecked")
+    @Test
+    void shouldPropagateTimeoutCreatingTopics() throws ExecutionException, InterruptedException, TimeoutException {
+        // Given
+        final CreateTopicsResult createTopicsResult = mock(CreateTopicsResult.class);
+        final KafkaFuture<Void> kafkaFuture = mock(KafkaFuture.class);
+        when(admin.createTopics(anyCollection())).thenReturn(createTopicsResult);
+        when(createTopicsResult.all()).thenReturn(kafkaFuture);
+        when(kafkaFuture.get(anyLong(), any(TimeUnit.class))).thenThrow(new TimeoutException("Kafka fell asleep"));
+
+        try (KroxyliciousTester tester = buildDefaultTester()) {
+
+            // When
+            assertThatThrownBy(() -> tester.createTopic(DEFAULT_CLUSTER))
+                    // Then
+                    .isInstanceOf(IllegalStateException.class)
+                    .hasMessage("Timed out creating topics on " + DEFAULT_CLUSTER);
+        }
+    }
+
+    @SuppressWarnings("unchecked")
+    @Test
+    void shouldPropagateErrorsDeletingTopics() throws ExecutionException, InterruptedException, TimeoutException {
+        // Given
+        allowCreateTopic(kroxyliciousClientsA, admin);
+
+        final KafkaFuture<Void> kafkaFuture = mock(KafkaFuture.class);
+        final DeleteTopicsResult deleteTopicsResult = mock(DeleteTopicsResult.class);
+        when(admin.deleteTopics(anyCollection())).thenReturn(deleteTopicsResult);
+        when(deleteTopicsResult.all()).thenReturn(kafkaFuture);
+        when(kafkaFuture.get(anyLong(), any(TimeUnit.class))).thenThrow(new ExecutionException("Kafka is nae working pal", new KafkaException("KaBoom")));
+
+        try (KroxyliciousTester tester = buildTester()) {
+            tester.createTopics(VIRTUAL_CLUSTER_A, TOPIC_COUNT);
+
+            // When
+            assertThatThrownBy(() -> tester.deleteTopics(VIRTUAL_CLUSTER_A))
+                    // Then
+                    .isInstanceOf(IllegalStateException.class)
+                    .hasMessage("Failed to delete topics on " + VIRTUAL_CLUSTER_A)
+                    .hasCauseInstanceOf(KafkaException.class);
+        }
+    }
+
+    @SuppressWarnings("unchecked")
+    @Test
+    void shouldPropagateTimeoutDeletingTopics() throws ExecutionException, InterruptedException, TimeoutException {
+        // Given
+        allowCreateTopic(kroxyliciousClientsA, admin);
+
+        final KafkaFuture<Void> kafkaFuture = mock(KafkaFuture.class);
+        final DeleteTopicsResult deleteTopicsResult = mock(DeleteTopicsResult.class);
+        when(admin.deleteTopics(anyCollection())).thenReturn(deleteTopicsResult);
+        when(deleteTopicsResult.all()).thenReturn(kafkaFuture);
+        when(kafkaFuture.get(anyLong(), any(TimeUnit.class))).thenThrow(new TimeoutException("Kafka fell asleep"));
+
+        try (KroxyliciousTester tester = buildTester()) {
+            tester.createTopics(VIRTUAL_CLUSTER_A, TOPIC_COUNT);
+
+            // When
+            assertThatThrownBy(() -> tester.deleteTopics(VIRTUAL_CLUSTER_A))
+                    // Then
+                    .isInstanceOf(IllegalStateException.class)
+                    .hasMessage("Timed out deleting topics on " + VIRTUAL_CLUSTER_A);
+        }
+    }
+
+    private void allowCreateTopic(KroxyliciousClients kroxyliciousClients, Admin admin) {
+        when(kroxyliciousClients.admin()).thenReturn(admin);
+        final CreateTopicsResult createTopicsResultA = mock(CreateTopicsResult.class);
+        when(admin.createTopics(anyCollection())).thenReturn(createTopicsResultA);
+        when(createTopicsResultA.all()).thenReturn(KafkaFuture.completedFuture(null));
+    }
+
+    @NonNull
+    private static Consumer<Map<String, Object>> assertSslConfiguration(String trustStorePath) {
+        return actual -> assertThat(actual)
+                .contains(
+                        Map.entry(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, SecurityProtocol.SSL.name()),
+                        Map.entry(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, trustStorePath))
+                .containsKey(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG);
+    }
+
+    @NonNull
+    private KroxyliciousTester buildDefaultTester() {
+        return new KroxyliciousTesterBuilder().setConfigurationBuilder(proxy(backingCluster))
+                .setKroxyliciousFactory(DefaultKroxyliciousTester::spawnProxy)
+                .setClientFactory(clientFactory)
+                .createDefaultKroxyliciousTester();
+    }
+
     @NonNull
-    private DefaultKroxyliciousTester buildDefaultTester() {
-        return new DefaultKroxyliciousTester(proxy(backingCluster),
-                DefaultKroxyliciousTester::spawnProxy,
-                clientFactory);
+    private KroxyliciousTester buildSecureTester(KeytoolCertificateGenerator keytoolCertificateGenerator) {
+        generateSecurityCert(keytoolCertificateGenerator);
+        final ConfigurationBuilder configurationBuilder = new ConfigurationBuilder();
+        var vcb = new VirtualClusterBuilder()
+                .withNewTargetCluster()
+                .withBootstrapServers(backingCluster)
+                .endTargetCluster()
+                .withNewTls()
+                .withNewKeyStoreKey()
+                .withStoreFile(keytoolCertificateGenerator.getKeyStoreLocation())
+                .withNewInlinePasswordStoreProvider(keytoolCertificateGenerator.getPassword())
+                .endKeyStoreKey()
+                .endTls()
+                .withClusterNetworkAddressConfigProvider(
+                        new ClusterNetworkAddressConfigProviderDefinitionBuilder(PortPerBrokerClusterNetworkAddressConfigProvider.class.getName())
+                                .withConfig("bootstrapAddress", DEFAULT_PROXY_BOOTSTRAP).build());
+        configurationBuilder
+                .addToVirtualClusters(TLS_CLUSTER, vcb.build());
+
+        return new KroxyliciousTesterBuilder().setConfigurationBuilder(configurationBuilder)
+                .setTrustStoreLocation(keytoolCertificateGenerator.getTrustStoreLocation())
+                .setTrustStorePassword(keytoolCertificateGenerator.getPassword())
+                .setKroxyliciousFactory(DefaultKroxyliciousTester::spawnProxy).setClientFactory(clientFactory).createDefaultKroxyliciousTester();
+    }
+
+    private static void generateSecurityCert(KeytoolCertificateGenerator keytoolCertificateGenerator) {
+        try {
+            keytoolCertificateGenerator.generateSelfSignedCertificateEntry("webmaster@example.com", "example.com", "Engineering", "kroxylicious.io", null, null, "NZ");
+        }
+        catch (GeneralSecurityException | IOException e) {
+            fail("unable to generate security certificate", e);
+        }
+    }
+
+    private KroxyliciousTester buildTester() {
+        return new KroxyliciousTesterBuilder()
+                .setConfigurationBuilder(proxy(backingCluster, VIRTUAL_CLUSTER_A, VIRTUAL_CLUSTER_B, VIRTUAL_CLUSTER_C))
+                .setKroxyliciousFactory(DefaultKroxyliciousTester::spawnProxy).setClientFactory(clientFactory).createDefaultKroxyliciousTester();
     }
 
-    private DefaultKroxyliciousTester buildTester() {
-        return new DefaultKroxyliciousTester(proxy(backingCluster, VIRTUAL_CLUSTER_A, VIRTUAL_CLUSTER_B, VIRTUAL_CLUSTER_C),
-                DefaultKroxyliciousTester::spawnProxy,
-                clientFactory);
+    public static <T> T argThat(Consumer<T> assertions) {
+        return MockitoHamcrest.argThat(new AssertionMatcher<>() {
+            @Override
+            public void assertion(T actual) throws AssertionError {
+                assertions.accept(actual);
+            }
+        });
     }
 }
\ No newline at end of file
diff --git a/kroxylicious-test-tools/src/test/java/io/kroxylicious/test/tester/KroxyliciousClientsTest.java b/kroxylicious-test-tools/src/test/java/io/kroxylicious/test/tester/KroxyliciousClientsTest.java
index df2dd023..938f2c6a 100644
--- a/kroxylicious-test-tools/src/test/java/io/kroxylicious/test/tester/KroxyliciousClientsTest.java
+++ b/kroxylicious-test-tools/src/test/java/io/kroxylicious/test/tester/KroxyliciousClientsTest.java
@@ -12,6 +12,7 @@ import java.util.Set;
 import java.util.function.Supplier;
 import java.util.stream.Stream;
 
+import org.apache.kafka.clients.CommonClientConfigs;
 import org.apache.kafka.clients.admin.Admin;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.MockConsumer;
@@ -28,10 +29,10 @@ import org.junit.jupiter.api.TestInstance;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
 import org.junit.jupiter.params.provider.MethodSource;
-import org.mockito.Mockito;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
 
 @TestInstance(TestInstance.Lifecycle.PER_CLASS)
@@ -41,22 +42,27 @@ class KroxyliciousClientsTest {
 
     @BeforeEach
     void setUp() {
-        kroxyliciousClients = new KroxyliciousClients("kroxylicious.example.com:9091", new KroxyliciousClients.ClientFactory() {
-            @Override
-            public Admin newAdmin(Map<String, Object> clientConfiguration) {
-                return Mockito.mock(Admin.class);
-            }
-
-            @Override
-            public <K, V> Consumer<K, V> newConsumer(Map<String, Object> clientConfiguration, Deserializer<K> keyDeserializer, Deserializer<V> valueDeserializer) {
-                return new MockConsumer<>(OffsetResetStrategy.LATEST);
-            }
-
-            @Override
-            public <K, V> Producer<K, V> newProducer(Map<String, Object> clientConfiguration, Serializer<K> keySerializer, Serializer<V> valueSerializer) {
-                return new MockProducer<>();
-            }
-        });
+        kroxyliciousClients = new KroxyliciousClients(Map.of(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, "kroxylicious.example.com:9091"),
+                new KroxyliciousClients.ClientFactory() {
+                    @Override
+                    public Admin newAdmin(Map<String, Object> clientConfiguration) {
+                        assertThat(clientConfiguration).contains(Map.entry(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, "kroxylicious.example.com:9091"));
+                        return mock(Admin.class);
+                    }
+
+                    @Override
+                    public <K, V> Consumer<K, V> newConsumer(Map<String, Object> clientConfiguration, Deserializer<K> keyDeserializer,
+                                                             Deserializer<V> valueDeserializer) {
+                        assertThat(clientConfiguration).contains(Map.entry(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, "kroxylicious.example.com:9091"));
+                        return new MockConsumer<>(OffsetResetStrategy.LATEST);
+                    }
+
+                    @Override
+                    public <K, V> Producer<K, V> newProducer(Map<String, Object> clientConfiguration, Serializer<K> keySerializer, Serializer<V> valueSerializer) {
+                        assertThat(clientConfiguration).contains(Map.entry(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, "kroxylicious.example.com:9091"));
+                        return new MockProducer<>();
+                    }
+                });
     }
 
     @ParameterizedTest
@@ -187,14 +193,14 @@ class KroxyliciousClientsTest {
     public Stream<Arguments> producers() {
         return Stream.of(Arguments.of((Supplier<Producer<String, String>>) () -> kroxyliciousClients.producer(),
                 (Supplier<Producer<String, String>>) () -> kroxyliciousClients.producer(Map.of("configKey", "configValue")),
-                (Supplier<Producer<String, String>>) () -> kroxyliciousClients.producer(new Serdes.StringSerde(), new Serdes.StringSerde(),
+                (Supplier<Producer<String, String>>) () -> kroxyliciousClients.producer(Serdes.String(), Serdes.String(),
                         Map.of("configKey", "configValue"))));
     }
 
     public Stream<Arguments> consumers() {
         return Stream.of(Arguments.of((Supplier<Consumer<String, String>>) () -> kroxyliciousClients.consumer(),
                 (Supplier<Consumer<String, String>>) () -> kroxyliciousClients.consumer(Map.of("configKey", "configValue")),
-                (Supplier<Consumer<String, String>>) () -> kroxyliciousClients.consumer(new Serdes.StringSerde(), new Serdes.StringSerde(),
+                (Supplier<Consumer<String, String>>) () -> kroxyliciousClients.consumer(Serdes.String(), Serdes.String(),
                         Map.of("configKey", "configValue"))));
     }
 }
\ No newline at end of file