Repository: kroxylicious/kroxylicious
Pull request #612 (2023-09-15 03:33:14)

Support optional config objects 
### Type of change

- Bugfix
- Enhancement / new feature

### Description

Allows Contributors to define if a filter actually requires configuration or not.

Additional Context
In fixing https://github.com/kroxylicious/kroxylicious/issues/530 it made sense to add a check if configuration was required, however that requires us to introduce the notion of required config. The initial answer was to add a check for configClass != BaseConfig.

@robobario then posed the question

> if they specify a specific subclass of BaseConfig as the configType, should we require it to be non-null?

To which the obvious answer is **no** filters should be able to support optional configuration see [RejectingCreateTopicFilter](https://github.com/kroxylicious/kroxylicious/blob/15c3d32a850ae92f82d063e81df4530e84c9cd6e/integrationtests/src/test/java/io/kroxylicious/proxy/filter/RejectingCreateTopicFilter.java#L32C1-L32C1) as a good example. Thus we need a way for Filter authors to signal if configuration is required or not. The Contributor is the natural home for that as it is fully under the control of the filter author. Adding it to the FilterDefinition would expose it to filter users which doesn't make sense.

The one perhaps controversial aspect of this PR is it validates filter configuration during proxy startup and fails startup if there are any configuration errors. While that may not be our long term posture it is consistent with how we do port validation.

Replaces https://github.com/kroxylicious/kroxylicious/pull/602 as the rebase after merging https://github.com/kroxylicious/kroxylicious/pull/608 was just too hairy. 


### 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]: 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=612)

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

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


k-wall: It seems a bit weird to special case the validation of filter configuration, while other "definable" objects aren't validated to instance creation. Should we give the responsibility to the *Definition classes?

So in the case of FilterDefinition:

```
    @JsonCreator
    public FilterDefinition {
        Objects.requireNonNull(type);
        ConfigurationDefinition definition = ContributionManager.INSTANCE.getDefinition(FilterContributor.class, type);
        if (definition.configurationRequired() && config != null) {
            throw new IllegalStateException("blah");
        }
    }
```

That way configuration problems would be detected uniformly and early.

There might be a funky way to plug into the Jackson deserialisation framework like this https://www.baeldung.com/java-object-validation-deserialization and somehow make it generic code.  It'd need some way of going from the type known to Jackson (e.g. `FilterDefinition`) to its contributor (e.g. `FilterContributor`).


SamBarker: > It seems a bit weird to special case the validation of filter configuration, while other "definable" objects aren't validated to instance creation
I agree it seems a bit strange, and I poked at making a common validator (when working #602 ) however I backed away to focus on the problem at hand and to keep the scope smaller. 

I think that trade off is still valid. This PR provides a fix for the filter configuration problem, then we can follow up with a more generic solution in 0.4.0.

> Should we give the responsibility to the *Definition classes? 
> There might be a funky way to plug into the Jackson deserialisation framework like this https://www.baeldung.com/java-object-validation-deserialization and somehow make it generic code.

Throwing in `*Definition` does effectively plug-in to Jackson. I did consider failing at that point as well however doing it there (AFICT) would mean we can't do partial loading. By that I mean we wouldn't be able to load a 3 valid Virtual clusters and log a failure for the 1 virtual cluster which is miss configured. Partial loading feels like quite an important property for us to support if we want the proxy to function in the ingress/gateway API spaces.

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

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

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

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

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

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

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

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

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



Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 788086da..fa2c0b70 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -5,7 +5,7 @@ Please enumerate **all user-facing** changes using format `<githib issue/pr numb
 ## 1.0.0
 
 ## 0.3.0
-
+* [#612](https://github.com/kroxylicious/kroxylicious/pull/612): [Breaking] Allow filter authors to declare when their filter requires configuration. Note this includes a backwards incompatible change to the contract of the `Contributor`. `getInstance` will now throw exceptions rather than returning `null` to mean there was a problem or this contributor does not know about the requested type.
 * [#608](https://github.com/kroxylicious/kroxylicious/pull/608): Improve the contributor API to allow it to express more properties about the configuration. This release deprecates `Contributor.getConfigType` in favour of `Contributor.getConfigDefinition`. It also removes the proliferation of ContributorManager classes by providing a single type which can handle all Contributors.
 * [#538](https://github.com/kroxylicious/kroxylicious/pull/538): Refactor FilterHandler and fix several bugs that would leave messages unflushed to client/broker.
 * [#531](https://github.com/kroxylicious/kroxylicious/pull/531): Simple Test Client now supports multi-RPC conversations with the server.
diff --git a/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/BaseContributor.java b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/BaseContributor.java
index 8e0c979a..4eaec366 100644
--- a/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/BaseContributor.java
+++ b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/BaseContributor.java
@@ -7,12 +7,15 @@ package io.kroxylicious.proxy.service;
 
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Objects;
 import java.util.function.BiFunction;
 import java.util.function.Function;
 import java.util.function.Supplier;
 
 import io.kroxylicious.proxy.config.BaseConfig;
 
+import edu.umd.cs.findbugs.annotations.NonNull;
+
 /**
  * A convenience base class for creating concrete contributor subclasses using a typesafe builder
  *
@@ -20,14 +23,14 @@ import io.kroxylicious.proxy.config.BaseConfig;
  */
 public abstract class BaseContributor<T, S extends Context> implements Contributor<T, S> {
 
-    private final Map<String, ContributionDetails<T, S>> typeNameToInstanceBuilder;
+    private final Map<String, ContributionDetails<T, S>> typeNameToContributorDetails;
 
     /**
      * Constructs and configures the contributor using the supplied {@code builder}.
      * @param builder builder
      */
     protected BaseContributor(BaseContributorBuilder<T, S> builder) {
-        typeNameToInstanceBuilder = builder.build();
+        typeNameToContributorDetails = builder.build();
     }
 
     /**
@@ -37,7 +40,7 @@ public abstract class BaseContributor<T, S extends Context> implements Contribut
      */
     @Override
     public boolean contributes(String typeName) {
-        return typeNameToInstanceBuilder.containsKey(typeName);
+        return typeNameToContributorDetails.containsKey(typeName);
     }
 
     /**
@@ -48,8 +51,8 @@ public abstract class BaseContributor<T, S extends Context> implements Contribut
      */
     @Override
     public ConfigurationDefinition getConfigDefinition(String typeName) {
-        if (typeNameToInstanceBuilder.containsKey(typeName)) {
-            return typeNameToInstanceBuilder.get(typeName).configurationDefinition();
+        if (contributes(typeName)) {
+            return typeNameToContributorDetails.get(typeName).configurationDefinition();
         }
         else {
             throw new IllegalArgumentException("No configuration definition registered for " + typeName);
@@ -66,8 +69,8 @@ public abstract class BaseContributor<T, S extends Context> implements Contribut
     @Override
     @Deprecated(forRemoval = true, since = "0.3.0")
     public Class<? extends BaseConfig> getConfigType(String typeName) {
-        if (typeNameToInstanceBuilder.containsKey(typeName)) {
-            return typeNameToInstanceBuilder.get(typeName).configurationDefinition().configurationType();
+        if (contributes(typeName)) {
+            return typeNameToContributorDetails.get(typeName).configurationDefinition().configurationType();
         }
         else {
             return null;
@@ -76,12 +79,20 @@ public abstract class BaseContributor<T, S extends Context> implements Contribut
 
     @Override
     public T getInstance(String typeName, S context) {
-        if (typeNameToInstanceBuilder.containsKey(typeName)) {
-            InstanceBuilder<T, S> instanceBuilder = typeNameToInstanceBuilder.get(typeName).instanceBuilder();
-            return instanceBuilder == null ? null : instanceBuilder.construct(context);
+        if (contributes(typeName)) {
+            final ContributionDetails<T, S> contributionDetails = typeNameToContributorDetails.get(typeName);
+            final boolean configurationRequired = contributionDetails.configurationDefinition().configurationRequired();
+            final boolean hasConfiguration = context.getConfig() != null;
+            if (!configurationRequired || hasConfiguration) {
+                InstanceBuilder<T, S> instanceBuilder = contributionDetails.instanceBuilder();
+                return Objects.requireNonNull(instanceBuilder.construct(context), "Tried to instantiate '" + typeName + "' but the Contributor returned null");
+            }
+            else {
+                throw new IllegalArgumentException("'" + typeName + "' requires configuration but none was supplied");
+            }
         }
         else {
-            return null;
+            throw new IllegalArgumentException("'" + typeName + "' is not a provided type");
         }
     }
 
@@ -138,7 +149,7 @@ public abstract class BaseContributor<T, S extends Context> implements Contribut
          * @param <T> the configuration concrete type
          */
         public <T extends BaseConfig> BaseContributorBuilder<L, D> add(String typeName, Class<T> configClass, Function<T, L> instanceFunction) {
-            return add(typeName, configClass, (context, config) -> instanceFunction.apply(config));
+            return add(typeName, configClass, (context, config) -> instanceFunction.apply(config), !Objects.equals(BaseConfig.class, configClass));
         }
 
         /**
@@ -147,15 +158,17 @@ public abstract class BaseContributor<T, S extends Context> implements Contribut
          * @param typeName service short name
          * @param configClass concrete type of configuration required by the service
          * @param instanceFunction function that constructs the service instance
+         * @param configurationRequired {@code true} if the contribution requires configuration. {@code false} if the contribution uses configuration for optional properties or to override defaults.
          * @return this
          * @param <T> the configuration concrete type
          */
-        public <T extends BaseConfig> BaseContributorBuilder<L, D> add(String typeName, Class<T> configClass, BiFunction<D, T, L> instanceFunction) {
+        public <T extends BaseConfig> BaseContributorBuilder<L, D> add(String typeName, Class<T> configClass, BiFunction<D, T, L> instanceFunction,
+                                                                       boolean configurationRequired) {
             if (typeNameToInstanceBuilder.containsKey(typeName)) {
                 throw new IllegalArgumentException(typeName + " already registered");
             }
             typeNameToInstanceBuilder.put(typeName,
-                    new ContributionDetails<>(new ConfigurationDefinition(configClass), InstanceBuilder.builder(configClass, instanceFunction)));
+                    new ContributionDetails<>(new ConfigurationDefinition(configClass, configurationRequired), InstanceBuilder.builder(configClass, instanceFunction)));
             return this;
         }
 
@@ -178,7 +191,7 @@ public abstract class BaseContributor<T, S extends Context> implements Contribut
          * @return this
          */
         public BaseContributorBuilder<L, D> add(String typeName, Function<D, L> instanceFunction) {
-            return add(typeName, BaseConfig.class, (context, config) -> instanceFunction.apply(context));
+            return add(typeName, BaseConfig.class, (context, config) -> instanceFunction.apply(context), false);
         }
 
         Map<String, ContributionDetails<L, D>> build() {
@@ -196,5 +209,5 @@ public abstract class BaseContributor<T, S extends Context> implements Contribut
         return new BaseContributorBuilder<>();
     }
 
-    protected record ContributionDetails<T, D extends Context>(ConfigurationDefinition configurationDefinition, InstanceBuilder<T, D> instanceBuilder) {}
-}
\ No newline at end of file
+    protected record ContributionDetails<T, D extends Context>(ConfigurationDefinition configurationDefinition, @NonNull InstanceBuilder<T, D> instanceBuilder) {}
+}
diff --git a/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/ConfigurationDefinition.java b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/ConfigurationDefinition.java
index 4579e388..ab07d262 100644
--- a/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/ConfigurationDefinition.java
+++ b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/service/ConfigurationDefinition.java
@@ -10,7 +10,11 @@ import io.kroxylicious.proxy.config.BaseConfig;
 
 /**
  * Defines the details of how Contributions interact with configuration for {@link BaseContributor}'s purposes.
+ * <p>
+ * configurationRequired allows filter authors to tell Kroxylicious if the filter must be supplied with a config object.
+ * If the filter uses configuration to supply optional properties, or it can provide sensible default values for the configuration then it should be set to false.
  *
  * @param configurationType defines the expected class for configuration objects
+ * @param configurationRequired {@code true} if the contribution requires a non-null configuration object. {@code false} if the contribution can tolerate a null configuration object
  */
-public record ConfigurationDefinition(Class<? extends BaseConfig> configurationType) {}
\ No newline at end of file
+public record ConfigurationDefinition(Class<? extends BaseConfig> configurationType, boolean configurationRequired) {}
\ No newline at end of file
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/KafkaProxy.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/KafkaProxy.java
index 2233aabb..7392265c 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/KafkaProxy.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/KafkaProxy.java
@@ -9,8 +9,10 @@ import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.Collectors;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -33,6 +35,7 @@ import io.netty.incubator.channel.uring.IOUringEventLoopGroup;
 import io.netty.incubator.channel.uring.IOUringServerSocketChannel;
 import io.netty.util.concurrent.Future;
 
+import io.kroxylicious.proxy.bootstrap.FilterChainFactory;
 import io.kroxylicious.proxy.config.Configuration;
 import io.kroxylicious.proxy.config.MicrometerDefinition;
 import io.kroxylicious.proxy.config.admin.AdminHttpConfiguration;
@@ -106,6 +109,12 @@ public final class KafkaProxy implements AutoCloseable {
 
         bindingOperationProcessor.start(plainServerBootstrap, tlsServerBootstrap);
 
+        final Set<String> invalidFilters = FilterChainFactory.validateFilterConfiguration(config.filters());
+        if (!invalidFilters.isEmpty()) {
+            throw new IllegalStateException(invalidFilters.stream()
+                    .collect(Collectors.joining(", ", "Missing required config for [", "]")));
+        }
+
         // TODO: startup/shutdown should return a completionstage
         CompletableFuture.allOf(virtualClusters.stream().map(vc -> endpointRegistry.registerVirtualCluster(vc).toCompletableFuture()).toArray(CompletableFuture[]::new))
                 .join();
@@ -238,4 +247,4 @@ public final class KafkaProxy implements AutoCloseable {
         }
     }
 
-}
+}
\ No newline at end of file
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/bootstrap/FilterChainFactory.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/bootstrap/FilterChainFactory.java
index 1eed9ec4..8e5e589f 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/bootstrap/FilterChainFactory.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/bootstrap/FilterChainFactory.java
@@ -5,8 +5,12 @@
  */
 package io.kroxylicious.proxy.bootstrap;
 
+import java.util.Collection;
 import java.util.List;
 import java.util.Objects;
+import java.util.Set;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
 
 import io.kroxylicious.proxy.config.Configuration;
 import io.kroxylicious.proxy.config.FilterDefinition;
@@ -25,8 +29,22 @@ public class FilterChainFactory {
     private final Configuration config;
 
     public FilterChainFactory(Configuration config) {
-        Objects.requireNonNull(config);
-        this.config = config;
+        this.config = Objects.requireNonNull(config);
+    }
+
+    /**
+     * Validate that a collection of FilterDefinitions contains all the required configuration to configure the filter.
+     * @param filterDefinitions the candidates to validate.
+     * @return the Set of filter names which fail validation.
+     */
+    public static Set<String> validateFilterConfiguration(Collection<FilterDefinition> filterDefinitions) {
+        if (filterDefinitions == null || filterDefinitions.isEmpty()) {
+            return Set.of();
+        }
+        return filterDefinitions.stream()
+                .filter(Predicate.not(FilterDefinition::isDefinitionValid))
+                .map(FilterDefinition::type)
+                .collect(Collectors.toSet());
     }
 
     /**
@@ -39,6 +57,7 @@ public class FilterChainFactory {
         if (filters == null || filters.isEmpty()) {
             return List.of();
         }
+        validateFilterConfiguration(filters);
         return filters
                 .stream()
                 .map(f -> ContributionManager.INSTANCE.getInstance(FilterContributor.class, f.type(), context.wrap(f.config())))
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/config/FilterDefinition.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/config/FilterDefinition.java
index ad1c7fd0..0492981c 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/config/FilterDefinition.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/config/FilterDefinition.java
@@ -8,14 +8,27 @@ package io.kroxylicious.proxy.config;
 import java.util.Objects;
 
 import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.annotation.JsonTypeInfo;
 import com.fasterxml.jackson.databind.annotation.JsonTypeIdResolver;
 
+import io.kroxylicious.proxy.filter.FilterContributor;
+import io.kroxylicious.proxy.service.ConfigurationDefinition;
+import io.kroxylicious.proxy.service.ContributionManager;
+
 public record FilterDefinition(@JsonProperty(required = true) String type,
                                @JsonTypeInfo(use = JsonTypeInfo.Id.NAME, include = JsonTypeInfo.As.EXTERNAL_PROPERTY, property = "type") @JsonTypeIdResolver(FilterConfigTypeIdResolver.class) BaseConfig config) {
     @JsonCreator
     public FilterDefinition {
         Objects.requireNonNull(type);
     }
-}
+
+    @JsonIgnore
+    public boolean isDefinitionValid() {
+        // We could consider allowing Filter authors to supply a config validation function `config -> config.property != null` via the configurationDefinition
+        // We would then be able to apply that validation here.
+        final ConfigurationDefinition configurationDefinition = ContributionManager.INSTANCE.getDefinition(FilterContributor.class, type);
+        return config != null || !configurationDefinition.configurationRequired();
+    }
+}
\ No newline at end of file


Test patch:
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/filter/TestFilterContributor.java b/integrationtests/src/test/java/io/kroxylicious/proxy/filter/TestFilterContributor.java
index f141ac78..e65d9d22 100644
--- a/integrationtests/src/test/java/io/kroxylicious/proxy/filter/TestFilterContributor.java
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/filter/TestFilterContributor.java
@@ -13,12 +13,12 @@ public class TestFilterContributor extends BaseContributor<Filter, FilterConstru
     public static final BaseContributorBuilder<Filter, FilterConstructContext> FILTERS = BaseContributor.<Filter, FilterConstructContext> builder()
             .add("FixedClientId", FixedClientIdFilter.FixedClientIdFilterConfig.class, FixedClientIdFilter::new)
             .add("ApiVersionsMarkingFilter", ApiVersionsMarkingFilter::new)
-            .add("RequestResponseMarking", RequestResponseMarkingFilter.RequestResponseMarkingFilterConfig.class, RequestResponseMarkingFilter::new)
+            .add("RequestResponseMarking", RequestResponseMarkingFilter.RequestResponseMarkingFilterConfig.class, RequestResponseMarkingFilter::new, true)
             .add("OutOfBandSend", OutOfBandSendFilter.OutOfBandSendFilterConfig.class, OutOfBandSendFilter::new)
             .add("CompositePrefixingFixedClientId", CompositePrefixingFixedClientIdFilterConfig.class, CompositePrefixingFixedClientIdFilter::new)
-            .add("RejectingCreateTopic", RejectingCreateTopicFilter.RejectingCreateTopicFilterConfig.class, RejectingCreateTopicFilter::new);
+            .add("RejectingCreateTopic", RejectingCreateTopicFilter.RejectingCreateTopicFilterConfig.class, RejectingCreateTopicFilter::new, false);
 
     public TestFilterContributor() {
         super(FILTERS);
     }
-}
+}
\ No newline at end of file
diff --git a/kroxylicious-api/src/test/java/io/kroxylicious/proxy/service/BaseContributorTest.java b/kroxylicious-api/src/test/java/io/kroxylicious/proxy/service/BaseContributorTest.java
index 27420552..6c463a94 100644
--- a/kroxylicious-api/src/test/java/io/kroxylicious/proxy/service/BaseContributorTest.java
+++ b/kroxylicious-api/src/test/java/io/kroxylicious/proxy/service/BaseContributorTest.java
@@ -7,12 +7,14 @@ package io.kroxylicious.proxy.service;
 
 import java.util.concurrent.atomic.AtomicReference;
 
+import org.assertj.core.api.ThrowableAssert;
 import org.junit.jupiter.api.Test;
 
 import io.kroxylicious.proxy.config.BaseConfig;
 
 import static io.kroxylicious.proxy.service.Context.wrap;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 class BaseContributorTest {
@@ -24,6 +26,48 @@ class BaseContributorTest {
     static class AnotherConfig extends BaseConfig {
     }
 
+    @Test
+    void shouldReturnTrueForConfigurationRequiredViaDefaulting() {
+        // Given
+        BaseContributor.BaseContributorBuilder<Long, Context> builder = BaseContributor.builder();
+        builder.add("one", LongConfig.class, longConfig -> 1L);
+        BaseContributor<Long, Context> baseContributor = new BaseContributor<>(builder) {
+        };
+
+        // When
+        boolean actual = baseContributor.getConfigDefinition("one").configurationRequired();
+
+        assertThat(actual).isTrue();
+    }
+
+    @Test
+    void shouldReturnTrueForExplicitlySpecifiedConfigurationRequired() {
+        // Given
+        BaseContributor.BaseContributorBuilder<Long, Context> builder = BaseContributor.builder();
+        builder.add("one", LongConfig.class, (longConfig, context) -> 1L, true);
+        BaseContributor<Long, Context> baseContributor = new BaseContributor<>(builder) {
+        };
+
+        // When
+        boolean actual = baseContributor.getConfigDefinition("one").configurationRequired();
+
+        assertThat(actual).isTrue();
+    }
+
+    @Test
+    void shouldReturnFalseFroOptionalRegisteredConfig() {
+        // Given
+        BaseContributor.BaseContributorBuilder<Long, Context> builder = BaseContributor.builder();
+        builder.add("one", LongConfig.class, (longConfig, context) -> 1L, false);
+        BaseContributor<Long, Context> baseContributor = new BaseContributor<>(builder) {
+        };
+
+        // When
+        boolean actual = baseContributor.getConfigDefinition("one").configurationRequired();
+
+        assertThat(actual).isFalse();
+    }
+
     @Test
     void testDefaultConfigClass() {
         BaseContributor.BaseContributorBuilder<Long, Context> builder = BaseContributor.builder();
@@ -80,6 +124,22 @@ class BaseContributorTest {
         assertThat(contextRef).hasValue(context);
     }
 
+    @Test
+    void testContextAndConfigFunctionWithNoConfig() {
+        BaseContributor.BaseContributorBuilder<Long, Context> builder = BaseContributor.builder();
+        AtomicReference<Context> contextRef = new AtomicReference<>();
+        builder.add("one", (context -> {
+            contextRef.set(context);
+            return 1L;
+        }));
+        BaseContributor<Long, Context> baseContributor = new BaseContributor<>(builder) {
+        };
+        Context context = wrap(null);
+        Long instance = baseContributor.getInstance("one", context);
+        assertThat(instance).isEqualTo(1L);
+        assertThat(contextRef).hasValue(context);
+    }
+
     @Test
     void testSpecifyingConfigType() {
         BaseContributor.BaseContributorBuilder<Long, Context> builder = BaseContributor.builder();
@@ -117,7 +177,7 @@ class BaseContributorTest {
         builder.add("fromBaseConfig", LongConfig.class, (context, baseConfig) -> {
             contextRef.set(context);
             return baseConfig.value;
-        });
+        }, true);
         BaseContributor<Long, Context> baseContributor = new BaseContributor<>(builder) {
         };
         Context context = wrap(new LongConfig());
@@ -160,8 +220,113 @@ class BaseContributorTest {
         };
 
         // When
-        assertThatThrownBy(() -> baseContributor.getConfigDefinition("fromBaseConfig")).isInstanceOf(IllegalArgumentException.class);
+        var throwableAssert = assertThatThrownBy(() -> baseContributor.getConfigDefinition("fromBaseConfig"));
+
+        // Then
+        throwableAssert.isInstanceOf(IllegalArgumentException.class);
+    }
+
+    @Test
+    void shouldFailIfConfigIsNullAndRequired() {
+        // Given
+        BaseContributor.BaseContributorBuilder<Long, Context> builder = BaseContributor.builder();
+        builder.add("requiresConfig", LongConfig.class, (baseConfig, context) -> 1L, true);
+        BaseContributor<Long, Context> baseContributor = new BaseContributor<>(builder) {
+        };
+
+        // When
+        var throwableAssert = assertThatThrownBy(() -> baseContributor.getInstance("requiresConfig", () -> null));
+
+        // Then
+        throwableAssert.isInstanceOf(IllegalArgumentException.class).hasMessage("'requiresConfig' requires configuration but none was supplied");
+    }
+
+    @Test
+    void shouldConstructInstanceIfConfigIsNullAndNotRequired() {
+        // Given
+        BaseContributor.BaseContributorBuilder<Long, Context> builder = BaseContributor.builder();
+        builder.add("noConfigRequired", LongConfig.class, (longConfig, context) -> 1L, false);
+        BaseContributor<Long, Context> baseContributor = new BaseContributor<>(builder) {
+        };
+
+        // When
+        final Long actualInstance = baseContributor.getInstance("noConfigRequired", () -> null);
+
+        // Then
+        assertThat(actualInstance).isNotNull();
+    }
+
+    @Test
+    void shouldConstructInstanceIfConfigTypeIsSpecifiedButNotRequiredAndConfigIsNull() {
+        // Given
+        BaseContributor.BaseContributorBuilder<Long, Context> builder = BaseContributor.builder();
+        builder.add("configOptional", LongConfig.class, (longConfig, context) -> 1L, false);
+        BaseContributor<Long, Context> baseContributor = new BaseContributor<>(builder) {
+        };
+
+        // When
+        final Long actualInstance = baseContributor.getInstance("configOptional", () -> null);
+
+        // Then
+        assertThat(actualInstance).isNotNull();
+    }
+
+    @Test
+    void shouldConstructInstanceIfConfigIsNotNullAndRequired() {
+        // Given
+        BaseContributor.BaseContributorBuilder<Long, Context> builder = BaseContributor.builder();
+        builder.add("configRequired", LongConfig.class, longConfig -> longConfig.value);
+        BaseContributor<Long, Context> baseContributor = new BaseContributor<>(builder) {
+        };
+
+        // When
+        final Long actualInstance = baseContributor.getInstance("configRequired", LongConfig::new);
+
+        // Then
+        assertThat(actualInstance).isNotNull();
+    }
+
+    @Test
+    void shouldConstructInstanceIfConfigIsNotNullAndNotRequired() {
+        // Given
+        BaseContributor.BaseContributorBuilder<Long, Context> builder = BaseContributor.builder();
+        builder.add("configOptional", () -> 1L);
+        BaseContributor<Long, Context> baseContributor = new BaseContributor<>(builder) {
+        };
+
+        // When
+        final Long actualInstance = baseContributor.getInstance("configOptional", () -> null);
+
+        // Then
+        assertThat(actualInstance).isNotNull();
+    }
+
+    @Test
+    void shouldThrowExceptionIfNoTypeRegistered() {
+        // Given
+        BaseContributor<Long, Context> baseContributor = new BaseContributor<>(BaseContributor.builder()) {
+        };
+
+        // When
+        final ThrowableAssert.ThrowingCallable throwingCallable = () -> baseContributor.getInstance("unknown", BaseConfig::new);
+
+        // Then
+        assertThatExceptionOfType(IllegalArgumentException.class).isThrownBy(throwingCallable).withMessage("'unknown' is not a provided type");
+    }
+
+    @Test
+    void shouldThrowExceptionIfRegisteredInstanceBuilderReturnsNull() {
+        // Given
+        BaseContributor.BaseContributorBuilder<Long, Context> builder = BaseContributor.builder();
+        builder.add("registered", (context) -> null);
+        BaseContributor<Long, Context> baseContributor = new BaseContributor<>(builder) {
+        };
+
+        // When
+        final ThrowableAssert.ThrowingCallable throwingCallable = () -> baseContributor.getInstance("registered", BaseConfig::new);
 
         // Then
+        assertThatExceptionOfType(NullPointerException.class).isThrownBy(throwingCallable)
+                .withMessage("Tried to instantiate 'registered' but the Contributor returned null");
     }
 }
\ No newline at end of file
diff --git a/kroxylicious/src/test/java/io/kroxylicious/proxy/KafkaProxyTest.java b/kroxylicious/src/test/java/io/kroxylicious/proxy/KafkaProxyTest.java
index 642a3a9f..937b163d 100644
--- a/kroxylicious/src/test/java/io/kroxylicious/proxy/KafkaProxyTest.java
+++ b/kroxylicious/src/test/java/io/kroxylicious/proxy/KafkaProxyTest.java
@@ -8,17 +8,42 @@ package io.kroxylicious.proxy;
 
 import java.util.stream.Stream;
 
+import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
 import org.junit.jupiter.params.provider.MethodSource;
 
 import io.kroxylicious.proxy.config.ConfigParser;
+import io.kroxylicious.proxy.config.Configuration;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.fail;
 
 class KafkaProxyTest {
 
+    @Test
+    void shouldFailToStartIfRequireFilterConfigIsMissing() throws Exception {
+        var config = """
+                    virtualClusters:
+                        demo1:
+                            targetCluster:
+                              bootstrap_servers: kafka.example:1234
+                            clusterNetworkAddressConfigProvider:
+                              type: PortPerBroker
+                              config:
+                                bootstrapAddress: localhost:9192
+                                brokerStartPort: 9193
+                                numberOfBrokerPorts: 2
+                    filters:
+                       - type: ProduceRequestTransformation
+                """;
+        try (var kafkaProxy = new KafkaProxy(new ConfigParser().parseConfiguration(config))) {
+            assertThatThrownBy(kafkaProxy::startup).isInstanceOf(IllegalStateException.class).hasMessage("Missing required config for [ProduceRequestTransformation]");
+        }
+    }
+
     public static Stream<Arguments> detectsConflictingPorts() {
         return Stream.of(Arguments.of("bootstrap port conflict", """
                 virtualClusters:
@@ -89,10 +114,12 @@ class KafkaProxyTest {
     @MethodSource
     void missingTls(String name, String config, String expectedMessage) throws Exception {
 
-        var illegalArgumentException = assertThrows(IllegalStateException.class, () -> {
-            try (var kafkaProxy = new KafkaProxy(new ConfigParser().parseConfiguration(config))) {
+        final Configuration parsedConfiguration = new ConfigParser().parseConfiguration(config);
+        var illegalStateException = assertThrows(IllegalStateException.class, () -> {
+            try (var ignored = new KafkaProxy(parsedConfiguration)) {
+                fail("The proxy started, but a failure was expected.");
             }
         });
-        assertThat(illegalArgumentException).hasStackTraceContaining(expectedMessage);
+        assertThat(illegalStateException).hasStackTraceContaining(expectedMessage);
     }
 }
\ No newline at end of file
diff --git a/kroxylicious/src/test/java/io/kroxylicious/proxy/bootstrap/FilterChainFactoryTest.java b/kroxylicious/src/test/java/io/kroxylicious/proxy/bootstrap/FilterChainFactoryTest.java
index 8b3d6486..a14ee34e 100644
--- a/kroxylicious/src/test/java/io/kroxylicious/proxy/bootstrap/FilterChainFactoryTest.java
+++ b/kroxylicious/src/test/java/io/kroxylicious/proxy/bootstrap/FilterChainFactoryTest.java
@@ -7,10 +7,12 @@
 package io.kroxylicious.proxy.bootstrap;
 
 import java.util.List;
+import java.util.Set;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 
 import org.assertj.core.api.ListAssert;
+import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
 import io.kroxylicious.proxy.config.Configuration;
@@ -28,6 +30,15 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
 
 class FilterChainFactoryTest {
 
+    private ScheduledExecutorService eventLoop;
+    private ExampleConfig config;
+
+    @BeforeEach
+    void setUp() {
+        eventLoop = Executors.newScheduledThreadPool(1);
+        config = new ExampleConfig();
+    }
+
     @Test
     void testNullFiltersInConfigResultsInEmptyList() {
         ScheduledExecutorService eventLoop = Executors.newScheduledThreadPool(1);
@@ -44,23 +55,14 @@ class FilterChainFactoryTest {
 
     @Test
     void testEmptyFiltersInConfigResultsInEmptyList() {
-        ScheduledExecutorService eventLoop = Executors.newScheduledThreadPool(1);
-        FilterChainFactory filterChainFactory = new FilterChainFactory(new Configuration(null, null, List.of(), null, true));
-        List<FilterAndInvoker> filters = filterChainFactory.createFilters(new NettyFilterContext(eventLoop));
-        assertNotNull(filters, "Filters list should not be null");
-        assertTrue(filters.isEmpty(), "Filters list should be empty");
+        assertFiltersCreated(List.of());
     }
 
     @Test
     void testCreateFilter() {
-        ScheduledExecutorService eventLoop = Executors.newScheduledThreadPool(1);
-        ExampleConfig config = new ExampleConfig();
-        FilterChainFactory filterChainFactory = new FilterChainFactory(
-                new Configuration(null, null, List.of(new FilterDefinition(TestFilterContributor.SHORT_NAME_A, config)), null, true));
-        NettyFilterContext context = new NettyFilterContext(eventLoop);
-        List<FilterAndInvoker> filters = filterChainFactory.createFilters(context);
-        assertThat(filters).isNotNull().hasSize(1).first().extracting(FilterAndInvoker::filter).isInstanceOfSatisfying(TestFilter.class, testFilter -> {
-            assertThat(testFilter.getShortName()).isEqualTo(TestFilterContributor.SHORT_NAME_A);
+        final ListAssert<FilterAndInvoker> listAssert = assertFiltersCreated(List.of(new FilterDefinition(TestFilterContributor.TYPE_NAME_A, config)));
+        listAssert.first().extracting(FilterAndInvoker::filter).isInstanceOfSatisfying(TestFilter.class, testFilter -> {
+            assertThat(testFilter.getShortName()).isEqualTo(TestFilterContributor.TYPE_NAME_A);
             assertThat(testFilter.getContext().getConfig()).isSameAs(config);
             assertThat(testFilter.getContext().executors().eventLoop()).isSameAs(eventLoop);
             assertThat(testFilter.getExampleConfig()).isSameAs(config);
@@ -69,27 +71,82 @@ class FilterChainFactoryTest {
 
     @Test
     void testCreateFilters() {
-        ScheduledExecutorService eventLoop = Executors.newScheduledThreadPool(1);
-        ExampleConfig config = new ExampleConfig();
-        FilterDefinition definitionA = new FilterDefinition(TestFilterContributor.SHORT_NAME_A, config);
-        FilterDefinition definitionB = new FilterDefinition(TestFilterContributor.SHORT_NAME_B, config);
-        List<FilterDefinition> filterDefinitions = List.of(definitionA, definitionB);
-        FilterChainFactory filterChainFactory = new FilterChainFactory(new Configuration(null, null, filterDefinitions, null, true));
-        NettyFilterContext context = new NettyFilterContext(eventLoop);
-        List<FilterAndInvoker> filters = filterChainFactory.createFilters(context);
-        ListAssert<FilterAndInvoker> listAssert = assertThat(filters).isNotNull().hasSize(2);
+        final ListAssert<FilterAndInvoker> listAssert = assertFiltersCreated(List.of(new FilterDefinition(TestFilterContributor.TYPE_NAME_A, config),
+                new FilterDefinition(TestFilterContributor.TYPE_NAME_B, config)));
         listAssert.element(0).extracting(FilterAndInvoker::filter).isInstanceOfSatisfying(TestFilter.class, testFilter -> {
-            assertThat(testFilter.getShortName()).isEqualTo(TestFilterContributor.SHORT_NAME_A);
+            assertThat(testFilter.getShortName()).isEqualTo(TestFilterContributor.TYPE_NAME_A);
             assertThat(testFilter.getContext().getConfig()).isSameAs(config);
             assertThat(testFilter.getContext().executors().eventLoop()).isSameAs(eventLoop);
             assertThat(testFilter.getExampleConfig()).isSameAs(config);
         });
         listAssert.element(1).extracting(FilterAndInvoker::filter).isInstanceOfSatisfying(TestFilter.class, testFilter -> {
-            assertThat(testFilter.getShortName()).isEqualTo(TestFilterContributor.SHORT_NAME_B);
+            assertThat(testFilter.getShortName()).isEqualTo(TestFilterContributor.TYPE_NAME_B);
             assertThat(testFilter.getContext().getConfig()).isSameAs(config);
             assertThat(testFilter.getContext().executors().eventLoop()).isSameAs(eventLoop);
             assertThat(testFilter.getExampleConfig()).isSameAs(config);
         });
     }
 
+    @Test
+    void shouldReturnInvalidFilterNameIfFilterRequiresConfigAndNoneIsSupplied() {
+        // Given
+        final List<FilterDefinition> filters = List.of(new FilterDefinition(TestFilterContributor.TYPE_NAME_A, config),
+                new FilterDefinition(TestFilterContributor.TYPE_NAME_B, null));
+
+        // When
+        final Set<String> invalidFilters = FilterChainFactory.validateFilterConfiguration(filters);
+
+        // Then
+        assertThat(invalidFilters).containsOnly(TestFilterContributor.TYPE_NAME_B);
+    }
+
+    @Test
+    void shouldReturnInvalidFilterNamesForAllFiltersWithoutRequiredConfig() {
+        // Given
+        final List<FilterDefinition> filters = List.of(new FilterDefinition(TestFilterContributor.TYPE_NAME_A, null),
+                new FilterDefinition(TestFilterContributor.TYPE_NAME_B, null),
+                new FilterDefinition(TestFilterContributor.OPTIONAL_CONFIG_FILTER, null));
+
+        // When
+        final Set<String> invalidFilters = FilterChainFactory.validateFilterConfiguration(filters);
+
+        // Then
+        assertThat(invalidFilters).containsOnly(TestFilterContributor.TYPE_NAME_A,
+                TestFilterContributor.TYPE_NAME_B);
+    }
+
+    @Test
+    void shouldPassValidationWhenAllFiltersHaveConfiguration() {
+        // Given
+        final List<FilterDefinition> filterDefinitions = List.of(new FilterDefinition(TestFilterContributor.TYPE_NAME_A, config),
+                new FilterDefinition(TestFilterContributor.TYPE_NAME_B, config));
+
+        // When
+        final Set<String> invalidFilters = FilterChainFactory.validateFilterConfiguration(filterDefinitions);
+
+        // Then
+        assertThat(invalidFilters).isEmpty();
+    }
+
+    @Test
+    void shouldPassValidationWhenFiltersWithOptionalConfigurationAreMissingConfiguration() {
+        // Given
+        final List<FilterDefinition> filterDefinitions = List.of(new FilterDefinition(TestFilterContributor.TYPE_NAME_A, config),
+                new FilterDefinition(TestFilterContributor.TYPE_NAME_B, config),
+                new FilterDefinition(TestFilterContributor.OPTIONAL_CONFIG_FILTER, null));
+
+        // When
+        final Set<String> invalidFilters = FilterChainFactory.validateFilterConfiguration(filterDefinitions);
+
+        // Then
+        assertThat(invalidFilters).isEmpty();
+    }
+
+    private ListAssert<FilterAndInvoker> assertFiltersCreated(List<FilterDefinition> filterDefinitions) {
+        FilterChainFactory filterChainFactory = new FilterChainFactory(new Configuration(null, null, filterDefinitions, null, true));
+        NettyFilterContext context = new NettyFilterContext(eventLoop);
+        List<FilterAndInvoker> filters = filterChainFactory.createFilters(context);
+        return assertThat(filters).isNotNull().hasSize(filterDefinitions.size());
+    }
+
 }
diff --git a/kroxylicious/src/test/java/io/kroxylicious/proxy/config/FilterDefinitionTest.java b/kroxylicious/src/test/java/io/kroxylicious/proxy/config/FilterDefinitionTest.java
new file mode 100644
index 00000000..68268b16
--- /dev/null
+++ b/kroxylicious/src/test/java/io/kroxylicious/proxy/config/FilterDefinitionTest.java
@@ -0,0 +1,66 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.proxy.config;
+
+import org.junit.jupiter.api.Test;
+
+import io.kroxylicious.proxy.internal.filter.ExampleConfig;
+
+import static io.kroxylicious.proxy.internal.filter.TestFilterContributor.OPTIONAL_CONFIG_FILTER;
+import static io.kroxylicious.proxy.internal.filter.TestFilterContributor.REQUIRED_CONFIG_FILTER;
+import static org.assertj.core.api.Assertions.assertThat;
+
+class FilterDefinitionTest {
+
+    @Test
+    void shouldFailValidationIfRequireConfigMissing() {
+        // Given
+        final FilterDefinition requiredConfig = new FilterDefinition(REQUIRED_CONFIG_FILTER, null);
+
+        // When
+        final boolean actual = requiredConfig.isDefinitionValid();
+
+        // Then
+        assertThat(actual).isFalse();
+    }
+
+    @Test
+    void shouldPassValidationIfRequireConfigSupplied() {
+        // Given
+        final FilterDefinition requiredConfig = new FilterDefinition(REQUIRED_CONFIG_FILTER, new ExampleConfig());
+
+        // When
+        final boolean actual = requiredConfig.isDefinitionValid();
+
+        // Then
+        assertThat(actual).isTrue();
+    }
+
+    @Test
+    void shouldPassValidationIfOptionalConfigSupplied() {
+        // Given
+        final FilterDefinition requiredConfig = new FilterDefinition(OPTIONAL_CONFIG_FILTER, new ExampleConfig());
+
+        // When
+        final boolean actual = requiredConfig.isDefinitionValid();
+
+        // Then
+        assertThat(actual).isTrue();
+    }
+
+    @Test
+    void shouldPassValidationIfOptionalConfigIsMissing() {
+        // Given
+        final FilterDefinition requiredConfig = new FilterDefinition(OPTIONAL_CONFIG_FILTER, null);
+
+        // When
+        final boolean actual = requiredConfig.isDefinitionValid();
+
+        // Then
+        assertThat(actual).isTrue();
+    }
+}
\ No newline at end of file
diff --git a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/clusternetworkaddressconfigprovider/TestClusterNetworkAddressConfigProviderContributor.java b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/clusternetworkaddressconfigprovider/TestClusterNetworkAddressConfigProviderContributor.java
index 1b5f6d38..ffd4b67f 100644
--- a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/clusternetworkaddressconfigprovider/TestClusterNetworkAddressConfigProviderContributor.java
+++ b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/clusternetworkaddressconfigprovider/TestClusterNetworkAddressConfigProviderContributor.java
@@ -34,7 +34,7 @@ public class TestClusterNetworkAddressConfigProviderContributor implements Clust
     @Override
     public ConfigurationDefinition getConfigDefinition(String shortName) {
         if (Objects.equals(shortName, SHORT_NAME)) {
-            return new ConfigurationDefinition(Config.class);
+            return new ConfigurationDefinition(Config.class, true);
         }
         return null;
     }
diff --git a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/filter/TestFilterContributor.java b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/filter/TestFilterContributor.java
index 99c4e143..19440801 100644
--- a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/filter/TestFilterContributor.java
+++ b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/filter/TestFilterContributor.java
@@ -12,13 +12,17 @@ import io.kroxylicious.proxy.filter.FilterContributor;
 import io.kroxylicious.proxy.service.BaseContributor;
 
 public class TestFilterContributor extends BaseContributor<Filter, FilterConstructContext> implements FilterContributor {
-    public static final String SHORT_NAME_A = "TEST1";
-    public static final String SHORT_NAME_B = "TEST2";
+    public static final String TYPE_NAME_A = "TEST1";
+    public static final String TYPE_NAME_B = "TEST2";
+    public static final String OPTIONAL_CONFIG_FILTER = "TEST3";
+    public static final String REQUIRED_CONFIG_FILTER = "TEST4";
     public static final BaseContributorBuilder<Filter, FilterConstructContext> FILTERS = BaseContributor.<Filter, FilterConstructContext> builder()
-            .add(SHORT_NAME_A, ExampleConfig.class, (context, exampleConfig) -> new TestFilter(SHORT_NAME_A, context, exampleConfig))
-            .add(SHORT_NAME_B, ExampleConfig.class, (context, exampleConfig) -> new TestFilter(SHORT_NAME_B, context, exampleConfig));
+            .add(TYPE_NAME_A, ExampleConfig.class, (context, exampleConfig) -> new TestFilter(TYPE_NAME_A, context, exampleConfig), true)
+            .add(TYPE_NAME_B, ExampleConfig.class, (context, exampleConfig) -> new TestFilter(TYPE_NAME_B, context, exampleConfig), true)
+            .add(REQUIRED_CONFIG_FILTER, ExampleConfig.class, (context, exampleConfig) -> new TestFilter(REQUIRED_CONFIG_FILTER, context, exampleConfig), true)
+            .add(OPTIONAL_CONFIG_FILTER, ExampleConfig.class, (context, exampleConfig) -> new TestFilter(OPTIONAL_CONFIG_FILTER, context, exampleConfig), false);
 
     public TestFilterContributor() {
         super(FILTERS);
     }
-}
+}
\ No newline at end of file
diff --git a/kroxylicious/src/test/java/io/kroxylicious/proxy/micrometer/TestMicrometerConfigurationHookContributor.java b/kroxylicious/src/test/java/io/kroxylicious/proxy/micrometer/TestMicrometerConfigurationHookContributor.java
index d905f286..e62b456c 100644
--- a/kroxylicious/src/test/java/io/kroxylicious/proxy/micrometer/TestMicrometerConfigurationHookContributor.java
+++ b/kroxylicious/src/test/java/io/kroxylicious/proxy/micrometer/TestMicrometerConfigurationHookContributor.java
@@ -38,7 +38,7 @@ public class TestMicrometerConfigurationHookContributor implements MicrometerCon
     @Override
     public ConfigurationDefinition getConfigDefinition(String shortName) {
         if (Objects.equals(shortName, SHORT_NAME)) {
-            return new ConfigurationDefinition(Config.class);
+            return new ConfigurationDefinition(Config.class, true);
         }
         else {
             return null;
diff --git a/kroxylicious/src/test/java/io/kroxylicious/proxy/service/ContributionManagerTest.java b/kroxylicious/src/test/java/io/kroxylicious/proxy/service/ContributionManagerTest.java
index 84c5ae76..1f3b5d3c 100644
--- a/kroxylicious/src/test/java/io/kroxylicious/proxy/service/ContributionManagerTest.java
+++ b/kroxylicious/src/test/java/io/kroxylicious/proxy/service/ContributionManagerTest.java
@@ -131,7 +131,7 @@ class ContributionManagerTest {
 
         @Override
         public ConfigurationDefinition getConfigDefinition(String shortName) {
-            return new ConfigurationDefinition(LongConfig.class);
+            return new ConfigurationDefinition(LongConfig.class, true);
         }
 
         @Override
@@ -168,7 +168,7 @@ class ContributionManagerTest {
 
         @Override
         public ConfigurationDefinition getConfigDefinition(String shortName) {
-            return new ConfigurationDefinition(configurationType);
+            return new ConfigurationDefinition(configurationType, true);
         }
 
         @Override