Repository: kroxylicious/kroxylicious
Pull request #309 (2023-05-08 23:25:12)

Add produce filter that can validate records meet some criteria
### Type of change

- Enhancement / new feature

### Description

Some organisations might want stronger guards against invalid records being produced to Kafka. The existing schema support is client side and there is no protection against some client coming along and pushing unexpected bytes to your topic, which could require manual intervention at the consumer if it cannot progress past the unexpected record.

If the filter finds any invalid records in a Produce request:

1. If the filter finds that all topic-partition data in a request has any invalid record in them, it generates a response marking all topic-partitions as failed with error and responds to the client without proxying.
2. If the request has a transactionalId, it generates a response marking all topic-partitions as failed with error and responds to the client without proxying.
3. If the user has specified that they do not want to forward partial produce data,i t generates a response marking all topic-partitions as failed with error and responds to the client without proxying.
4. If the user has specified that they do want to forward partial produce data then all topic-partitions containing only valid records are sent to the proxied broker. When the response from the broker is intercepted, all failed topic-partitions are augmented into the response with corresponding error codes and messages.

This feature is built on the assumption that one topic should have one validator. We use Rules to determine which validation to apply to a given topic. By iterating through the rules in the order they are declared, checking if a topic's name matches each rule, we locate the first matching rule. If no rule is matched this way we can configure a default validation rule to be applied. Note: the rules application is calculated once per topic and then cached in a map.

For example, this configuration has a rule for topic "one" and topic "two", plus a customised default rule:

```
- type: ProduceValidator
  config:
    defaultRule:
      valueRule:
        allowNulls: true
        allowEmpty: false
    rules:
    - topicNames:
      - one
      valueRule:
        syntacticallyCorrectJson:
          validateObjectKeysUnique: true
    - topicNames:
      - two
      valueRule:
        allowNulls: false
        allowEmpty: false
```

The first validation implemented is JSON syntax validation, checking if the value is well-formed JSON. The user can optionally enable it to validate the uniqueness of keys in JSON objects.

This is built with flexibility in the YAML configuration and implementation in mind, so that further validations can be plugged in later. For example it would be great to integrate with Schema registries to check that the data is encoded using a schema.

robobario: I've been hacking on apicurio integration and it has exposed some things that might be worth changing about the APIs.

When you use Apicurio's kafka-client integration (using default apicurio producer config), it uses the message headers to pass schema groupId/artifactId/version. The BytebufValidator` interface currently only knows about a buffer and length to read from it. So we either need to pass down the headers, or we could find some alternative like removing the `BytebufValidator` interface and use `RecordValidator` to do the same work. So we would make `RecordValidators` that can configurably target the key or value. This feels better to me as we will have the entire record available for future Validators to work with.
k-wall: How will the filter handle compressed records?  I think by default it should be prepared to uncompress the record so it can be subjected to the validations.  This should probably be configurable too.
robobario: > How will the filter handle compressed records? I think by default it should be prepared to uncompress the record so it can be subjected to the validations. This should probably be configurable too.

Compression is at the `RecordBatch` level, it transparently decompresses them when you iterate over the Records of the batch. So we should be working with decompressed bytes already. I'll check this tomorrow though.
robobario: Have confirmed all JsonSyntaxValidationIT cases pass if you set `compression.type=gzip`

Patch:
diff --git a/kroxylicious-schema-validation/pom.xml b/kroxylicious-schema-validation/pom.xml
new file mode 100644
index 00000000..e0795915
--- /dev/null
+++ b/kroxylicious-schema-validation/pom.xml
@@ -0,0 +1,85 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+    Copyright Kroxylicious Authors.
+
+    Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <parent>
+        <groupId>io.kroxylicious</groupId>
+        <artifactId>kroxylicious-parent</artifactId>
+        <version>0.2.0-SNAPSHOT</version>
+        <relativePath>../pom.xml</relativePath>
+    </parent>
+
+    <artifactId>kroxylicious-schema-validation</artifactId>
+    <packaging>jar</packaging>
+
+    <name>Kroxylicious Schema Validation</name>
+    <description>Kroxylicious Filters and other plugins required to validate the schema of messages
+    </description>
+
+    <dependencies>
+        <dependency>
+            <groupId>io.kroxylicious</groupId>
+            <artifactId>kroxylicious-api</artifactId>
+            <version>${kroxyliciousApi.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>io.kroxylicious</groupId>
+            <artifactId>kroxylicious-filter-api</artifactId>
+            <version>${kroxyliciousApi.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-databind</artifactId>
+        </dependency>
+        <!-- Test dependencies -->
+        <dependency>
+            <groupId>com.google.guava</groupId>
+            <artifactId>guava</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.dataformat</groupId>
+            <artifactId>jackson-dataformat-yaml</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.logging.log4j</groupId>
+            <artifactId>log4j-slf4j2-impl</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-params</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.assertj</groupId>
+            <artifactId>assertj-core</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-core</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>com.flipkart.zjsonpatch</groupId>
+            <artifactId>zjsonpatch</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+</project>
diff --git a/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/ProduceRequestValidationFilterContributor.java b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/ProduceRequestValidationFilterContributor.java
new file mode 100644
index 00000000..b224b361
--- /dev/null
+++ b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/ProduceRequestValidationFilterContributor.java
@@ -0,0 +1,31 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.kroxylicious.proxy.filter.schema;
+
+import io.kroxylicious.proxy.filter.FilterContributor;
+import io.kroxylicious.proxy.filter.KrpcFilter;
+import io.kroxylicious.proxy.filter.schema.config.ValidationConfig;
+import io.kroxylicious.proxy.filter.schema.validation.request.ProduceRequestValidator;
+import io.kroxylicious.proxy.service.BaseContributor;
+
+/**
+ * Contributor for request validation filters
+ */
+public class ProduceRequestValidationFilterContributor extends BaseContributor<KrpcFilter> implements FilterContributor {
+
+    private static final BaseContributorBuilder<KrpcFilter> FILTERS = BaseContributor.<KrpcFilter> builder()
+            .add("ProduceValidator", ValidationConfig.class, (config) -> {
+                ProduceRequestValidator validator = ProduceValidationFilterBuilder.build(config);
+                return new ProduceValidationFilter(config.isForwardPartialRequests(), validator);
+            });
+
+    /**
+     * Constructor (called via ${@link java.util.ServiceLoader})
+     */
+    public ProduceRequestValidationFilterContributor() {
+        super(FILTERS);
+    }
+}
diff --git a/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/ProduceValidationFilter.java b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/ProduceValidationFilter.java
new file mode 100644
index 00000000..a78cb4fb
--- /dev/null
+++ b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/ProduceValidationFilter.java
@@ -0,0 +1,185 @@
+/*
+ * 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.filter.schema;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.kafka.common.message.ProduceRequestData;
+import org.apache.kafka.common.message.ProduceResponseData;
+import org.apache.kafka.common.message.RequestHeaderData;
+import org.apache.kafka.common.message.ResponseHeaderData;
+import org.apache.kafka.common.protocol.Errors;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.kroxylicious.proxy.filter.KrpcFilterContext;
+import io.kroxylicious.proxy.filter.ProduceRequestFilter;
+import io.kroxylicious.proxy.filter.ProduceResponseFilter;
+import io.kroxylicious.proxy.filter.schema.validation.request.ProduceRequestValidationResult;
+import io.kroxylicious.proxy.filter.schema.validation.request.ProduceRequestValidator;
+import io.kroxylicious.proxy.filter.schema.validation.topic.PartitionValidationResult;
+import io.kroxylicious.proxy.filter.schema.validation.topic.RecordValidationFailure;
+import io.kroxylicious.proxy.filter.schema.validation.topic.TopicValidationResult;
+
+/**
+ * A Filter that is intended to validate some criteria about each topic-partition, preventing
+ * invalid data being sent on to the broker. The response will contain an {@link Errors#INVALID_RECORD}
+ * error code and an error message for each invalid topic partition.
+ * <p>
+ * Optionally supports forwarding partial data. If the request is non-transactional, the filter
+ * can be configured to forward any valid topic-partitions to the broker, filtering out invalid ones.
+ * Then when the ProduceResponse is intercepted then it can augment in failure messages for the invalid
+ * topic-partitions.
+ * </p>
+ * <p>
+ * Note: if all the topic partitions are invalid (or the request is transactional), a response is sent
+ * back to the client without forwarding anything upstream, with all topic-partitions failed.
+ * </p>
+ */
+public class ProduceValidationFilter implements ProduceRequestFilter, ProduceResponseFilter {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(ProduceValidationFilter.class);
+    private final boolean forwardPartialRequests;
+    private final ProduceRequestValidator validator;
+    private final Map<Integer, ProduceRequestValidationResult> correlatedResults = new HashMap<>();
+
+    /**
+     * Construct a new ProduceValidationFilter
+     * @param forwardPartialRequests whether to forward valid topic-partitions if some other topic-partition is invalid (transactional requests are never forwarded if any topic-partition invalid)
+     * @param validator validator to test ProduceRequests with
+     */
+    public ProduceValidationFilter(boolean forwardPartialRequests, ProduceRequestValidator validator) {
+        if (validator == null) {
+            throw new IllegalArgumentException("validator is null");
+        }
+        this.forwardPartialRequests = forwardPartialRequests;
+        this.validator = validator;
+    }
+
+    @Override
+    public void onProduceRequest(RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context) {
+        ProduceRequestValidationResult result = validator.validateRequest(request);
+        if (result.isAnyTopicPartitionInvalid()) {
+            handleInvalidTopicPartitions(header, request, context, result);
+        }
+        else {
+            context.forwardRequest(header, request);
+        }
+    }
+
+    private void handleInvalidTopicPartitions(RequestHeaderData header, ProduceRequestData request, KrpcFilterContext context, ProduceRequestValidationResult result) {
+        if (result.isAllTopicPartitionsInvalid()) {
+            LOGGER.debug("all topic-partitions for request contained invalid data: {}", result);
+            ProduceResponseData response = invalidateEntireRequest(request, result);
+            context.forwardResponse(response);
+        }
+        // do not forward partial produce data if request is transactional because the whole produce must eventually succeed or fail together
+        else if (request.transactionalId() == null && forwardPartialRequests) {
+            LOGGER.debug("some topic-partitions contained invalid data: {}, forwarding valid topic-partitions", result);
+            request.topicData().removeIf(topicProduceData -> result.isAllPartitionsInvalid(topicProduceData.name()));
+            for (ProduceRequestData.TopicProduceData topicDatum : request.topicData()) {
+                topicDatum.partitionData().removeIf(partitionProduceData -> !result.isPartitionValid(topicDatum.name(), partitionProduceData.index()));
+            }
+            correlatedResults.put(header.correlationId(), result);
+            context.forwardRequest(header, request);
+        }
+        else {
+            LOGGER.debug("some topic-partitions for transactional request with id: {}, contained invalid data: {}, invalidation entire request",
+                    request.transactionalId(), result);
+            ProduceResponseData response = invalidateEntireRequest(request, result);
+            context.forwardResponse(response);
+        }
+    }
+
+    private static ProduceResponseData invalidateEntireRequest(ProduceRequestData request, ProduceRequestValidationResult produceRequestValidationResult) {
+        ProduceResponseData response = new ProduceResponseData();
+        ProduceResponseData.TopicProduceResponseCollection responseCollection = new ProduceResponseData.TopicProduceResponseCollection();
+        request.topicData().forEach(topicProduceData -> {
+            String topicName = topicProduceData.name();
+            TopicValidationResult topicValidationResult = produceRequestValidationResult.topicResult(topicName);
+            ProduceResponseData.TopicProduceResponse newElement = createInvalidatedTopicProduceResponse(topicName, topicProduceData, topicValidationResult);
+            responseCollection.add(newElement);
+        });
+        response.setResponses(responseCollection);
+        return response;
+    }
+
+    private static ProduceResponseData.TopicProduceResponse createInvalidatedTopicProduceResponse(String topicName, ProduceRequestData.TopicProduceData topicProduceData,
+                                                                                                  TopicValidationResult topicValidationResult) {
+        ProduceResponseData.TopicProduceResponse response = new ProduceResponseData.TopicProduceResponse();
+        response.setName(topicName);
+        List<ProduceResponseData.PartitionProduceResponse> responses = topicProduceData.partitionData().stream().map(partitionProduceData -> {
+            PartitionValidationResult partitionResult = topicValidationResult.getPartitionResult(partitionProduceData.index());
+            return createInvalidatedPartitionProduceResponse(partitionProduceData, partitionResult);
+        }).toList();
+        response.setPartitionResponses(responses);
+        return response;
+    }
+
+    private static ProduceResponseData.PartitionProduceResponse createInvalidatedPartitionProduceResponse(ProduceRequestData.PartitionProduceData partitionProduceData,
+                                                                                                          PartitionValidationResult partitionResult) {
+        ProduceResponseData.PartitionProduceResponse produceResponse = new ProduceResponseData.PartitionProduceResponse();
+        produceResponse.setIndex(partitionProduceData.index());
+        produceResponse.setErrorCode(Errors.INVALID_RECORD.code());
+        if (partitionResult.allRecordsValid()) {
+            produceResponse.setErrorMessage("Invalid record in another topic-partition caused whole ProduceRequest to be invalidated");
+        }
+        else {
+            for (RecordValidationFailure recordValidationFailure : partitionResult.recordValidationFailures()) {
+                produceResponse.recordErrors().add(new ProduceResponseData.BatchIndexAndErrorMessage().setBatchIndex(recordValidationFailure.invalidIndex())
+                        .setBatchIndexErrorMessage(recordValidationFailure.errorMessage()));
+            }
+            produceResponse.setErrorMessage(toErrorString(partitionResult.recordValidationFailures()));
+        }
+        return produceResponse;
+    }
+
+    private static String toErrorString(List<RecordValidationFailure> failures) {
+        String failString = failures.stream().findFirst().map(RecordValidationFailure::errorMessage).orElse("Failure List Empty");
+        return "Records in batch were invalid: [" + failString + "]";
+    }
+
+    @Override
+    public void onProduceResponse(ResponseHeaderData header, ProduceResponseData response, KrpcFilterContext context) {
+        ProduceRequestValidationResult produceRequestValidationResult = correlatedResults.remove(header.correlationId());
+        if (produceRequestValidationResult != null) {
+            LOGGER.debug("augmenting invalid topic-partition details into response: {}", produceRequestValidationResult);
+            augmentResponseWithInvalidTopicPartitions(response, produceRequestValidationResult);
+            context.forwardResponse(response);
+        }
+        else {
+            context.forwardResponse(response);
+        }
+    }
+
+    private void augmentResponseWithInvalidTopicPartitions(ProduceResponseData response, ProduceRequestValidationResult produceRequestValidationResult) {
+        produceRequestValidationResult.topicsWithInvalidPartitions().forEach(topicWithInvalidPartitions -> {
+            ProduceResponseData.TopicProduceResponse topicProduceResponse = response.responses().find(topicWithInvalidPartitions.topicName());
+            if (topicProduceResponse == null) {
+                topicProduceResponse = new ProduceResponseData.TopicProduceResponse();
+                topicProduceResponse.setName(topicWithInvalidPartitions.topicName());
+                response.responses().add(topicProduceResponse);
+            }
+            augmentTopicProduceResponse(topicWithInvalidPartitions, topicProduceResponse);
+        });
+    }
+
+    private static void augmentTopicProduceResponse(TopicValidationResult topicWithInvalidPartitions, ProduceResponseData.TopicProduceResponse topicProduceResponse) {
+        topicWithInvalidPartitions.invalidPartitions().forEach(partitionValidationResult -> {
+            ProduceResponseData.PartitionProduceResponse response = new ProduceResponseData.PartitionProduceResponse();
+            response.setIndex(partitionValidationResult.index());
+            for (RecordValidationFailure recordValidationFailure : partitionValidationResult.recordValidationFailures()) {
+                response.recordErrors().add(new ProduceResponseData.BatchIndexAndErrorMessage().setBatchIndex(recordValidationFailure.invalidIndex())
+                        .setBatchIndexErrorMessage(recordValidationFailure.errorMessage()));
+            }
+            response.setErrorCode(Errors.INVALID_RECORD.code());
+            response.setErrorMessage(toErrorString(partitionValidationResult.recordValidationFailures()));
+            topicProduceResponse.partitionResponses().add(response);
+        });
+    }
+}
diff --git a/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/ProduceValidationFilterBuilder.java b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/ProduceValidationFilterBuilder.java
new file mode 100644
index 00000000..34b4b903
--- /dev/null
+++ b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/ProduceValidationFilterBuilder.java
@@ -0,0 +1,59 @@
+/*
+ * 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.filter.schema;
+
+import io.kroxylicious.proxy.filter.schema.config.BytebufValidation;
+import io.kroxylicious.proxy.filter.schema.config.RecordValidationRule;
+import io.kroxylicious.proxy.filter.schema.config.ValidationConfig;
+import io.kroxylicious.proxy.filter.schema.validation.bytebuf.BytebufValidator;
+import io.kroxylicious.proxy.filter.schema.validation.bytebuf.BytebufValidators;
+import io.kroxylicious.proxy.filter.schema.validation.record.KeyAndValueRecordValidator;
+import io.kroxylicious.proxy.filter.schema.validation.request.ProduceRequestValidator;
+import io.kroxylicious.proxy.filter.schema.validation.request.RoutingProduceRequestValidator;
+import io.kroxylicious.proxy.filter.schema.validation.topic.TopicValidator;
+import io.kroxylicious.proxy.filter.schema.validation.topic.TopicValidators;
+
+/**
+ * Builds from configuration objects to a ProduceRequestValidator
+ */
+public class ProduceValidationFilterBuilder {
+
+    private ProduceValidationFilterBuilder() {
+
+    }
+
+    /**
+     * Build a ProduceRequestValidator from configuration
+     * @param config configuration
+     * @return a ProduceRequestValidator
+     */
+    public static ProduceRequestValidator build(ValidationConfig config) {
+        RoutingProduceRequestValidator.RoutingProduceRequestValidatorBuilder builder = RoutingProduceRequestValidator.builder();
+        config.getRules().forEach(rule -> builder.appendValidatorForTopicPattern(rule.getTopicNames(), toValidatorWithNullHandling(rule)));
+        RecordValidationRule defaultRule = config.getDefaultRule();
+        TopicValidator defaultValidator = defaultRule == null ? TopicValidators.allValid() : toValidatorWithNullHandling(defaultRule);
+        builder.setDefaultValidator(defaultValidator);
+        return builder.build();
+    }
+
+    private static TopicValidator toValidatorWithNullHandling(RecordValidationRule validationRule) {
+        BytebufValidator keyValidator = validationRule.getKeyRule().map(ProduceValidationFilterBuilder::getBytebufValidator).orElse(BytebufValidators.allValid());
+        BytebufValidator valueValidator = validationRule.getValueRule().map(ProduceValidationFilterBuilder::getBytebufValidator).orElse(BytebufValidators.allValid());
+        return TopicValidators.perRecordValidator(KeyAndValueRecordValidator.keyAndValueValidator(keyValidator, valueValidator));
+    }
+
+    private static BytebufValidator getBytebufValidator(BytebufValidation validation) {
+        BytebufValidator innerValidator = toValidator(validation);
+        return BytebufValidators.nullEmptyValidator(validation.isAllowNulls(), validation.isAllowEmpty(), innerValidator);
+    }
+
+    private static BytebufValidator toValidator(BytebufValidation valueRule) {
+        return valueRule.getSyntacticallyCorrectJsonConfig().map(config -> BytebufValidators.jsonSyntaxValidator(config.isValidateObjectKeysUnique()))
+                .orElse(BytebufValidators.allValid());
+    }
+
+}
diff --git a/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/config/BytebufValidation.java b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/config/BytebufValidation.java
new file mode 100644
index 00000000..aa298161
--- /dev/null
+++ b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/config/BytebufValidation.java
@@ -0,0 +1,88 @@
+/*
+ * 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.filter.schema.config;
+
+import java.util.Objects;
+import java.util.Optional;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+/**
+ * Configuration for validating a Bytebuffer holding a value.
+ */
+public class BytebufValidation {
+    private final SyntacticallyCorrectJsonConfig syntacticallyCorrectJsonConfig;
+    private final boolean allowNulls;
+    private final boolean allowEmpty;
+
+    /**
+     * Create a new BytebufValidation
+     * @param syntacticallyCorrectJsonConfig optional configuration, if non-null indicates ByteBuffer should contain syntactically correct JSON
+     * @param allowNulls whether a null byte-buffer should be considered valid
+     * @param allowEmpty whether an empty byte-buffer should be considered valid
+     */
+    @JsonCreator
+    public BytebufValidation(@JsonProperty("syntacticallyCorrectJson") SyntacticallyCorrectJsonConfig syntacticallyCorrectJsonConfig,
+                             @JsonProperty(value = "allowNulls", defaultValue = "true") Boolean allowNulls,
+                             @JsonProperty(value = "allowEmpty", defaultValue = "false") Boolean allowEmpty) {
+        this.syntacticallyCorrectJsonConfig = syntacticallyCorrectJsonConfig;
+        this.allowNulls = allowNulls == null || allowNulls;
+        this.allowEmpty = allowEmpty != null && allowEmpty;
+    }
+
+    /**
+     * Get syntactically correct json config
+     * @return optional containing syntacticallyCorrectJsonConfig if non-null, empty otherwise
+     */
+    public Optional<SyntacticallyCorrectJsonConfig> getSyntacticallyCorrectJsonConfig() {
+        return Optional.ofNullable(syntacticallyCorrectJsonConfig);
+    }
+
+    /**
+     * Are buffers valid if they are null on the ${@link org.apache.kafka.common.record.Record}
+     * @return allowNulls
+     */
+    public boolean isAllowNulls() {
+        return allowNulls;
+    }
+
+    /**
+     * Are buffers valid if they are empty (non-null, 0 length) on the ${@link org.apache.kafka.common.record.Record}
+     * @return allowEmpty
+     */
+    public boolean isAllowEmpty() {
+        return allowEmpty;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        BytebufValidation that = (BytebufValidation) o;
+        return allowNulls == that.allowNulls && allowEmpty == that.allowEmpty && Objects.equals(syntacticallyCorrectJsonConfig,
+                that.syntacticallyCorrectJsonConfig);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(syntacticallyCorrectJsonConfig, allowNulls, allowEmpty);
+    }
+
+    @Override
+    public String toString() {
+        return "BytebufValidation{" +
+                "syntacticallyCorrectJsonConfig=" + syntacticallyCorrectJsonConfig +
+                ", allowNulls=" + allowNulls +
+                ", allowEmpty=" + allowEmpty +
+                '}';
+    }
+}
diff --git a/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/config/RecordValidationRule.java b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/config/RecordValidationRule.java
new file mode 100644
index 00000000..d65d5402
--- /dev/null
+++ b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/config/RecordValidationRule.java
@@ -0,0 +1,83 @@
+/*
+ * 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.filter.schema.config;
+
+import java.util.Objects;
+import java.util.Optional;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+/**
+ * Describes an optional key validation rule and an optional value validation rule which will
+ * be applied to a ${@link org.apache.kafka.common.record.Record} to validate its contents.
+ */
+public class RecordValidationRule {
+
+    /**
+     * rule to apply to record key
+     */
+    protected final BytebufValidation keyRule;
+
+    /**
+     * rule to apply to record value
+     */
+    protected final BytebufValidation valueRule;
+
+    /**
+     * Construct new RecordValidationRule
+     * @param keyRule optional validation to apply to Record's key
+     * @param valueRule optional validation to apply to Record's value
+     */
+    @JsonCreator
+    public RecordValidationRule(@JsonProperty(value = "keyRule") BytebufValidation keyRule,
+                                @JsonProperty(value = "valueRule") BytebufValidation valueRule) {
+        this.keyRule = keyRule;
+        this.valueRule = valueRule;
+    }
+
+    /**
+     * get optional key rule
+     * @return optional containing key rule if non-null, else empty optional
+     */
+    public Optional<BytebufValidation> getKeyRule() {
+        return Optional.ofNullable(keyRule);
+    }
+
+    /**
+     * get value rule
+     * @return optional containing value rule if non-null, else empty optional
+     */
+    public Optional<BytebufValidation> getValueRule() {
+        return Optional.ofNullable(valueRule);
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        RecordValidationRule that = (RecordValidationRule) o;
+        return Objects.equals(keyRule, that.keyRule) && Objects.equals(valueRule, that.valueRule);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(keyRule, valueRule);
+    }
+
+    @Override
+    public String toString() {
+        return "RecordValidationRule{" +
+                "keyRule=" + keyRule +
+                ", valueRule=" + valueRule +
+                '}';
+    }
+}
diff --git a/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/config/SyntacticallyCorrectJsonConfig.java b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/config/SyntacticallyCorrectJsonConfig.java
new file mode 100644
index 00000000..c5d83c77
--- /dev/null
+++ b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/config/SyntacticallyCorrectJsonConfig.java
@@ -0,0 +1,60 @@
+/*
+ * 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.filter.schema.config;
+
+import java.util.Objects;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+/**
+ * Configuration for validating a component ByteBuffer of a ${@link org.apache.kafka.common.record.Record} is syntactically correct JSON.
+ */
+public class SyntacticallyCorrectJsonConfig {
+    private final boolean validateObjectKeysUnique;
+
+    /**
+     * Construct SyntacticallyCorrectJsonConfig
+     * @param validateObjectKeysUnique whether we expect the Object keys in the JSON to be unique
+     */
+    @JsonCreator
+    public SyntacticallyCorrectJsonConfig(@JsonProperty(value = "validateObjectKeysUnique", defaultValue = "false") Boolean validateObjectKeysUnique) {
+        this.validateObjectKeysUnique = validateObjectKeysUnique != null && validateObjectKeysUnique;
+    }
+
+    /**
+     * Do we expect the Object keys in the JSON to be unique
+     * @return true if we want to validate that the Object keys to be unique
+     */
+    public boolean isValidateObjectKeysUnique() {
+        return validateObjectKeysUnique;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        SyntacticallyCorrectJsonConfig that = (SyntacticallyCorrectJsonConfig) o;
+        return validateObjectKeysUnique == that.validateObjectKeysUnique;
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(validateObjectKeysUnique);
+    }
+
+    @Override
+    public String toString() {
+        return "SyntacticallyCorrectJsonConfig{" +
+                "validateObjectKeysUnique=" + validateObjectKeysUnique +
+                '}';
+    }
+}
diff --git a/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/config/TopicMatchingRecordValidationRule.java b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/config/TopicMatchingRecordValidationRule.java
new file mode 100644
index 00000000..ef7b0865
--- /dev/null
+++ b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/config/TopicMatchingRecordValidationRule.java
@@ -0,0 +1,72 @@
+/*
+ * 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.filter.schema.config;
+
+import java.util.Objects;
+import java.util.Set;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+/**
+ * Configuration that links a topic name pattern with some record validation rules.
+ */
+public class TopicMatchingRecordValidationRule extends RecordValidationRule {
+
+    private final Set<String> topicNames;
+
+    /**
+     * Construct a new TopicPatternRecordValidationRule
+     * @param topicNames required topic names topics with these names are eligible to have this rule applied to them
+     * @param keyRule optional validation to apply to Record's key
+     * @param valueRule optional validation to apply to Record's value
+     */
+    @JsonCreator
+    public TopicMatchingRecordValidationRule(@JsonProperty(value = "topicNames") Set<String> topicNames,
+                                             @JsonProperty(value = "keyRule") BytebufValidation keyRule,
+                                             @JsonProperty(value = "valueRule") BytebufValidation valueRule) {
+        super(keyRule, valueRule);
+        this.topicNames = topicNames == null ? Set.of() : topicNames;
+    }
+
+    /**
+     * Get topic name pattern that this rule is eligible to apply to
+     * @return topic name pattern
+     */
+    public Set<String> getTopicNames() {
+        return topicNames;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        if (!super.equals(o)) {
+            return false;
+        }
+        TopicMatchingRecordValidationRule that = (TopicMatchingRecordValidationRule) o;
+        return Objects.equals(topicNames, that.topicNames);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(super.hashCode(), topicNames);
+    }
+
+    @Override
+    public String toString() {
+        return "TopicMatchingRecordValidationRule{" +
+                "topicNames=" + topicNames +
+                ", keyRule=" + keyRule +
+                ", valueRule=" + valueRule +
+                '}';
+    }
+}
diff --git a/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/config/ValidationConfig.java b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/config/ValidationConfig.java
new file mode 100644
index 00000000..69b57c05
--- /dev/null
+++ b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/config/ValidationConfig.java
@@ -0,0 +1,99 @@
+/*
+ * 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.filter.schema.config;
+
+import java.util.List;
+import java.util.Objects;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import io.kroxylicious.proxy.config.BaseConfig;
+
+/**
+ * Configuration for Produce Request validation. Contains a description of the rules for validating
+ * the data for all topic-partitions with a ProduceRequest and how to handle partial failures (where
+ * some topic-partitions are valid and others are invalid within a single ProduceRequest)
+ */
+public class ValidationConfig extends BaseConfig {
+
+    /**
+     * If this is enabled then the proxy will (for non-transactional requests):
+     * 1. filter out invalid topic-partitions from the ProduceRequest
+     * 2. forward the filtered request up the chain
+     * 3. intercept the produce response and augment in failure details for the invalid topic-partitions
+     */
+    private final boolean forwardPartialRequests;
+    private final List<TopicMatchingRecordValidationRule> rules;
+    private final RecordValidationRule defaultRule;
+
+    /**
+     * Construct a new ValidationConfig
+     * @param forwardPartialRequests describes whether partial ProduceRequest data should be forwarded to the broker (for non-transactional requests)
+     * @param rules describes a list of rules, associating topics with some validation to be applied to produce data for that topic
+     * @param defaultRule the default validation rule to be applied when no rule is matched for a topic within a ProduceRequest
+     */
+    @JsonCreator
+    public ValidationConfig(@JsonProperty(value = "forwardPartialRequests", defaultValue = "false") Boolean forwardPartialRequests,
+                            @JsonProperty("rules") List<TopicMatchingRecordValidationRule> rules,
+                            @JsonProperty("defaultRule") RecordValidationRule defaultRule) {
+        this.forwardPartialRequests = forwardPartialRequests != null && forwardPartialRequests;
+        this.rules = rules;
+        this.defaultRule = defaultRule;
+    }
+
+    /**
+     * Get the rules
+     * @return rules
+     */
+    public List<TopicMatchingRecordValidationRule> getRules() {
+        return rules;
+    }
+
+    /**
+     * is forwarding partial requests enabled?
+     * @return true if forwarding partial requests enabled
+     */
+    public boolean isForwardPartialRequests() {
+        return forwardPartialRequests;
+    }
+
+    /**
+     * get default rule
+     * @return default rule (not null)
+     */
+    public RecordValidationRule getDefaultRule() {
+        return defaultRule;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        ValidationConfig that = (ValidationConfig) o;
+        return forwardPartialRequests == that.forwardPartialRequests && Objects.equals(rules, that.rules) && Objects.equals(defaultRule,
+                that.defaultRule);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(forwardPartialRequests, rules, defaultRule);
+    }
+
+    @Override
+    public String toString() {
+        return "ValidationConfig{" +
+                "forwardPartialRequests=" + forwardPartialRequests +
+                ", rules=" + rules +
+                ", defaultRule=" + defaultRule +
+                '}';
+    }
+}
diff --git a/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/Result.java b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/Result.java
new file mode 100644
index 00000000..af89f264
--- /dev/null
+++ b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/Result.java
@@ -0,0 +1,20 @@
+/*
+ * 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.filter.schema.validation;
+
+/**
+ * Result for a validation
+ * @param valid whether the input was valid
+ * @param errorMessage error message that should be supplied when input is invalid
+ */
+public record Result(boolean valid, String errorMessage) {
+
+    /**
+     * valid result
+     */
+    public static Result VALID = new Result(true, null);
+}
diff --git a/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/bytebuf/AllValidBytebufValidator.java b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/bytebuf/AllValidBytebufValidator.java
new file mode 100644
index 00000000..072c954f
--- /dev/null
+++ b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/bytebuf/AllValidBytebufValidator.java
@@ -0,0 +1,21 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.proxy.filter.schema.validation.bytebuf;
+
+import java.nio.ByteBuffer;
+
+import org.apache.kafka.common.record.Record;
+
+import io.kroxylicious.proxy.filter.schema.validation.Result;
+
+class AllValidBytebufValidator implements BytebufValidator {
+
+    @Override
+    public Result validate(ByteBuffer buffer, int length, Record record, boolean isKey) {
+        return Result.VALID;
+    }
+}
diff --git a/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/bytebuf/BytebufValidator.java b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/bytebuf/BytebufValidator.java
new file mode 100644
index 00000000..f786b017
--- /dev/null
+++ b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/bytebuf/BytebufValidator.java
@@ -0,0 +1,35 @@
+/*
+ * 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.filter.schema.validation.bytebuf;
+
+import java.nio.ByteBuffer;
+
+import org.apache.kafka.common.record.Record;
+
+import io.kroxylicious.proxy.filter.schema.validation.Result;
+
+/**
+ * Used to validate a ByteBuffer against some criteria
+ */
+public interface BytebufValidator {
+
+    /**
+     * Validate a ByteBuffer.
+     * <p>
+     * You can expect this ByteBuffer instance to not be re-used outside
+     * this validator so you don't have to mark/reset it after use. Though it will
+     * likely be backed by a shared buffer so do not write to it.
+     * </p>
+     *
+     * @param buffer the buffer containing data
+     * @param length the length of the value in the buffer (buffer may contain more data after the value)
+     * @param record the record the buffer was extracted from
+     * @param isKey true if the buffer is the key of the record, false if it is the value of the record
+     * @return a valid result if the buffer is valid
+     */
+    Result validate(ByteBuffer buffer, int length, Record record, boolean isKey);
+}
diff --git a/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/bytebuf/BytebufValidators.java b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/bytebuf/BytebufValidators.java
new file mode 100644
index 00000000..e1f025aa
--- /dev/null
+++ b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/bytebuf/BytebufValidators.java
@@ -0,0 +1,47 @@
+/*
+ * 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.filter.schema.validation.bytebuf;
+
+/**
+ * Static factory methods for creating/getting ${@link BytebufValidator} instances
+ */
+public class BytebufValidators {
+
+    private BytebufValidators() {
+
+    }
+
+    private static final AllValidBytebufValidator ALL_VALID = new AllValidBytebufValidator();
+
+    /**
+     * get validator that validates all {@link java.nio.ByteBuffer}s
+     * @return validator
+     */
+    public static BytebufValidator allValid() {
+        return ALL_VALID;
+    }
+
+    /**
+     * get validator that validates null/empty {@link java.nio.ByteBuffer}s and then delegates non-null/non-empty {@link java.nio.ByteBuffer}s to a delegate
+     * @param nullValid are null buffers valide\
+     * @param emptyValid are empty buffers valid
+     * @param delegate delegate to call if buffer is non-null/non-empty
+     * @return validator
+     */
+    public static BytebufValidator nullEmptyValidator(boolean nullValid, boolean emptyValid, BytebufValidator delegate) {
+        return new NullEmptyBytebufValidator(nullValid, emptyValid, delegate);
+    }
+
+    /**
+     * get validator that validates if a non-null/non-empty buffer contains syntactically correct JSON
+     * @param validateObjectKeysUnique optionally check if JSON Objects contain unique keys
+     * @return validator
+     */
+    public static BytebufValidator jsonSyntaxValidator(boolean validateObjectKeysUnique) {
+        return new JsonSyntaxBytebufValidator(validateObjectKeysUnique);
+    }
+}
diff --git a/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/bytebuf/JsonSyntaxBytebufValidator.java b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/bytebuf/JsonSyntaxBytebufValidator.java
new file mode 100644
index 00000000..aa260b8b
--- /dev/null
+++ b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/bytebuf/JsonSyntaxBytebufValidator.java
@@ -0,0 +1,58 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.proxy.filter.schema.validation.bytebuf;
+
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+
+import org.apache.kafka.common.record.Record;
+import org.apache.kafka.common.utils.ByteBufferInputStream;
+
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import io.kroxylicious.proxy.filter.schema.validation.Result;
+
+/**
+ * Checks if a Record's value is well-formed JSON, optionally checking if
+ * Object keys are unique. Object key uniqueness is not a hard requirement
+ * in the spec but some consumer implementations may expect them to be unique.
+ */
+class JsonSyntaxBytebufValidator implements BytebufValidator {
+    private final boolean validateObjectKeysUnique;
+
+    static final ObjectMapper mapper = new ObjectMapper().enable(DeserializationFeature.FAIL_ON_TRAILING_TOKENS);
+
+    JsonSyntaxBytebufValidator(boolean validateObjectKeysUnique) {
+        this.validateObjectKeysUnique = validateObjectKeysUnique;
+    }
+
+    @Override
+    public Result validate(ByteBuffer buffer, int size, Record record, boolean isKey) {
+        if (buffer == null) {
+            throw new IllegalArgumentException("buffer is null");
+        }
+        if (size < 1) {
+            throw new IllegalArgumentException("size is less than 1");
+        }
+        try (InputStream inputStream = new ByteBufferInputStream(buffer);
+                JsonParser parser = mapper.getFactory().createParser(inputStream)) {
+            if (validateObjectKeysUnique) {
+                parser.enable(JsonParser.Feature.STRICT_DUPLICATE_DETECTION);
+            }
+            while (parser.nextToken() != null) {
+            }
+            return Result.VALID;
+        }
+        catch (Exception e) {
+            String message = "value was not syntactically correct JSON" + (e.getMessage() != null ? ": " + e.getMessage() : "");
+            return new Result(false, message);
+        }
+    }
+
+}
diff --git a/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/bytebuf/NullEmptyBytebufValidator.java b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/bytebuf/NullEmptyBytebufValidator.java
new file mode 100644
index 00000000..fc6ff64c
--- /dev/null
+++ b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/bytebuf/NullEmptyBytebufValidator.java
@@ -0,0 +1,44 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.proxy.filter.schema.validation.bytebuf;
+
+import java.nio.ByteBuffer;
+
+import org.apache.kafka.common.record.Record;
+
+import io.kroxylicious.proxy.filter.schema.validation.Result;
+
+class NullEmptyBytebufValidator implements BytebufValidator {
+
+    private final boolean nullValid;
+    private final boolean emptyValid;
+    private final BytebufValidator delegate;
+
+    NullEmptyBytebufValidator(boolean nullValid, boolean emptyValid, BytebufValidator delegate) {
+        if (delegate == null) {
+            throw new IllegalArgumentException("delegate is null");
+        }
+        this.nullValid = nullValid;
+        this.emptyValid = emptyValid;
+        this.delegate = delegate;
+    }
+
+    @Override
+    public Result validate(ByteBuffer buffer, int length, Record record, boolean isKey) {
+        if (buffer == null) {
+            return result(nullValid, "Null buffer invalid");
+        }
+        else if (length == 0) {
+            return result(emptyValid, "Empty buffer invalid");
+        }
+        return delegate.validate(buffer, length, record, isKey);
+    }
+
+    private Result result(boolean allowed, String message) {
+        return allowed ? Result.VALID : new Result(false, message);
+    }
+}
diff --git a/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/record/KeyAndValueRecordValidator.java b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/record/KeyAndValueRecordValidator.java
new file mode 100644
index 00000000..b63b8940
--- /dev/null
+++ b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/record/KeyAndValueRecordValidator.java
@@ -0,0 +1,56 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.proxy.filter.schema.validation.record;
+
+import org.apache.kafka.common.record.Record;
+
+import io.kroxylicious.proxy.filter.schema.validation.Result;
+import io.kroxylicious.proxy.filter.schema.validation.bytebuf.BytebufValidator;
+
+/**
+ * Returns an invalid {@link Result} if either the key or value returns
+ * an invalid result from a delegate validator
+ */
+public class KeyAndValueRecordValidator implements RecordValidator {
+
+    private final BytebufValidator keyValidator;
+    private final BytebufValidator valueValidator;
+
+    private KeyAndValueRecordValidator(BytebufValidator keyValidator, BytebufValidator valueValidator) {
+        if (keyValidator == null) {
+            throw new IllegalArgumentException("keyValidator was null");
+        }
+        if (valueValidator == null) {
+            throw new IllegalArgumentException("valueValidator was null");
+        }
+        this.keyValidator = keyValidator;
+        this.valueValidator = valueValidator;
+    }
+
+    @Override
+    public Result validate(Record record) {
+        Result keyValid = keyValidator.validate(record.key(), record.keySize(), record, true);
+        if (!keyValid.valid()) {
+            return new Result(false, "Key was invalid: " + keyValid.errorMessage());
+        }
+        Result valueValid = valueValidator.validate(record.value(), record.valueSize(), record, false);
+        if (!valueValid.valid()) {
+            return new Result(false, "Value was invalid: " + valueValid.errorMessage());
+        }
+        return Result.VALID;
+    }
+
+    /**
+     * Get a RecordValidator that invalidates a record if either it's key or value fails validation with their respective validator
+     * @param keyValidator validator for the record's key
+     * @param valueValidator validator for the record's value
+     * @return validator
+     */
+    public static RecordValidator keyAndValueValidator(BytebufValidator keyValidator, BytebufValidator valueValidator) {
+        return new KeyAndValueRecordValidator(keyValidator, valueValidator);
+    }
+}
diff --git a/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/record/RecordValidator.java b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/record/RecordValidator.java
new file mode 100644
index 00000000..f18924a2
--- /dev/null
+++ b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/record/RecordValidator.java
@@ -0,0 +1,24 @@
+/*
+ * 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.filter.schema.validation.record;
+
+import org.apache.kafka.common.record.Record;
+
+import io.kroxylicious.proxy.filter.schema.validation.Result;
+
+/**
+ * Validator for individual ${@link Record}s
+ */
+public interface RecordValidator {
+
+    /**
+     * Validate the record
+     * @param record the record to be validated
+     * @return a Result describing if the record is valid and any failure message/exception if it is not.
+     */
+    Result validate(Record record);
+}
diff --git a/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/request/ProduceRequestValidationResult.java b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/request/ProduceRequestValidationResult.java
new file mode 100644
index 00000000..932a4c54
--- /dev/null
+++ b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/request/ProduceRequestValidationResult.java
@@ -0,0 +1,89 @@
+/*
+ * 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.filter.schema.validation.request;
+
+import java.util.Map;
+import java.util.stream.Stream;
+
+import io.kroxylicious.proxy.filter.schema.validation.topic.PartitionValidationResult;
+import io.kroxylicious.proxy.filter.schema.validation.topic.TopicValidationResult;
+
+/**
+ * The result of validating an entire ${@link org.apache.kafka.common.message.ProduceRequestData}. Contains
+ * validation results for each topic in the request.
+ * @param topicValidationResults results per topic, key is topicName
+ */
+public record ProduceRequestValidationResult(Map<String, TopicValidationResult> topicValidationResults) {
+
+    /**
+     * Is any topic partition invalid
+     * @return true if any topic-partitions were invalid
+     */
+    public boolean isAnyTopicPartitionInvalid() {
+        return topicValidationResults.values().stream().anyMatch(TopicValidationResult::isAnyPartitionInvalid);
+    }
+
+    /**
+     * Are all topic partitions invalid
+     * @return true if all topic-partitions were invalid
+     */
+    public boolean isAllTopicPartitionsInvalid() {
+        return topicValidationResults.values().stream().allMatch(TopicValidationResult::isAllPartitionsInvalid);
+    }
+
+    /**
+     * Are all topic partitions invalid for a topic
+     * @param topicName topic name
+     * @return true if all partitions for topicName are invalid
+     */
+    public boolean isAllPartitionsInvalid(String topicName) {
+        TopicValidationResult topicValidationResult = topicValidationResults.get(topicName);
+        if (topicValidationResult == null) {
+            return false;
+        }
+        else {
+            return topicValidationResult.isAllPartitionsInvalid();
+        }
+    }
+
+    /**
+     * Is a topic-partition valid
+     * @param topicName name of the topic
+     * @param partitionIndex index of the partition
+     * @return true if the topic-partition is valid
+     * @throws IllegalStateException if the topicName or partitionIndex has no recorded result, we should have some outcome for every topic-partition
+     */
+    public boolean isPartitionValid(String topicName, int partitionIndex) {
+        TopicValidationResult topicValidationResult = topicValidationResults.get(topicName);
+        if (topicValidationResult == null) {
+            throw new IllegalStateException("topicValidationResults should contain a result for all topics in the request, failed topic: " + topicName);
+        }
+        PartitionValidationResult partitionValidationResult = topicValidationResult.getPartitionResult(partitionIndex);
+        if (partitionValidationResult == null) {
+            throw new IllegalStateException(
+                    "partitionValidationResult should contain a result for all topics in the request, failed topic: " + topicName + ", partition" + partitionIndex);
+        }
+        return partitionValidationResult.allRecordsValid();
+    }
+
+    /**
+     * Get all topics with invalid partitions
+     * @return stream of topic results containing invalid partitions
+     */
+    public Stream<TopicValidationResult> topicsWithInvalidPartitions() {
+        return topicValidationResults.values().stream().filter(TopicValidationResult::isAnyPartitionInvalid);
+    }
+
+    /**
+     * Get topic result for a topic
+     * @param topicName name of topic
+     * @return result
+     */
+    public TopicValidationResult topicResult(String topicName) {
+        return topicValidationResults.get(topicName);
+    }
+}
\ No newline at end of file
diff --git a/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/request/ProduceRequestValidator.java b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/request/ProduceRequestValidator.java
new file mode 100644
index 00000000..80ada923
--- /dev/null
+++ b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/request/ProduceRequestValidator.java
@@ -0,0 +1,23 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.proxy.filter.schema.validation.request;
+
+import org.apache.kafka.common.message.ProduceRequestData;
+
+/**
+ * Validate that all Records in a Produce Request are valid and return a result
+ * describing which records were invalid.
+ */
+public interface ProduceRequestValidator {
+
+    /**
+     * Validate a request
+     * @param request the request
+     * @return result describing a validation outcome for all topic partitions and details of records that failed validation
+     */
+    ProduceRequestValidationResult validateRequest(ProduceRequestData request);
+}
diff --git a/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/request/RoutingProduceRequestValidator.java b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/request/RoutingProduceRequestValidator.java
new file mode 100644
index 00000000..81b5c06d
--- /dev/null
+++ b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/request/RoutingProduceRequestValidator.java
@@ -0,0 +1,120 @@
+/*
+ * 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.filter.schema.validation.request;
+
+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.function.Predicate;
+import java.util.stream.Collectors;
+
+import org.apache.kafka.common.message.ProduceRequestData;
+
+import io.kroxylicious.proxy.filter.schema.validation.topic.TopicValidationResult;
+import io.kroxylicious.proxy.filter.schema.validation.topic.TopicValidator;
+import io.kroxylicious.proxy.filter.schema.validation.topic.TopicValidators;
+
+/**
+ * A validator that can apply a different validation to each topic contained in
+ * a single produce request. It routes each topic's data to a validator instance using
+ * a list of rules. It applies the respective validator to each topic and then collects
+ * and returns a result describing the Results for all topics.
+ * <p>
+ * If no rule is matched for a topic, then a (configurable) default validator will be
+ * applied to the data for that topic.
+ * </p>
+ */
+public class RoutingProduceRequestValidator implements ProduceRequestValidator {
+
+    private final List<RoutingRule> rules;
+    private final TopicValidator defaultValidator;
+    private final Map<String, TopicValidator> cache = new HashMap<>();
+
+    private record RoutingRule(Predicate<String> topicPredicate, TopicValidator validator) {
+
+    }
+
+    private RoutingProduceRequestValidator(List<RoutingRule> rules, TopicValidator defaultValidator) {
+        if (rules == null) {
+            throw new IllegalArgumentException("rules is null");
+        }
+        if (defaultValidator == null) {
+            throw new IllegalArgumentException("defaultValidator is null");
+        }
+        this.rules = rules;
+        this.defaultValidator = defaultValidator;
+    }
+
+    @Override
+    public ProduceRequestValidationResult validateRequest(ProduceRequestData request) {
+        Map<String, TopicValidationResult> collect = request.topicData().stream().collect(
+                Collectors.toMap(ProduceRequestData.TopicProduceData::name, topicProduceData -> getTopicValidator(topicProduceData).validateTopicData(topicProduceData)));
+        return new ProduceRequestValidationResult(collect);
+    }
+
+    private TopicValidator getTopicValidator(ProduceRequestData.TopicProduceData topicProduceData) {
+        return cache.computeIfAbsent(topicProduceData.name(), topicName -> {
+            Optional<RoutingRule> first = rules.stream().filter(routingRule -> routingRule.topicPredicate().test(topicName)).findFirst();
+            return first.map(RoutingRule::validator).orElse(defaultValidator);
+        });
+    }
+
+    /**
+     * builder
+     * @return builder
+     */
+    public static RoutingProduceRequestValidatorBuilder builder() {
+        return new RoutingProduceRequestValidatorBuilder();
+    }
+
+    /**
+     * Builder for RoutingProduceRequestValidator
+     */
+    public static class RoutingProduceRequestValidatorBuilder {
+        private TopicValidator defaultValidator = TopicValidators.allValid();
+        private final List<RoutingRule> routingRules = new ArrayList<>();
+
+        private RoutingProduceRequestValidatorBuilder() {
+        }
+
+        /**
+         * set the default validator to be used if no RoutingRule matches a topic
+         * @param validator default validator
+         * @return this RoutingProduceRequestValidatorBuilder
+         */
+        public RoutingProduceRequestValidatorBuilder setDefaultValidator(TopicValidator validator) {
+            if (validator == null) {
+                throw new IllegalArgumentException("attempted to set a null default validator");
+            }
+            this.defaultValidator = validator;
+            return this;
+        }
+
+        /**
+         * append a validator rule for a topic pattern (note order matters, rules are applied to topics in append order)
+         * @param topicNames topic names
+         * @param validator validator to be applied if pattern matches topic
+         * @return this RoutingProduceRequestValidatorBuilder
+         */
+        public RoutingProduceRequestValidatorBuilder appendValidatorForTopicPattern(Set<String> topicNames, TopicValidator validator) {
+            routingRules.add(new RoutingRule(topicNames::contains, validator));
+            return this;
+        }
+
+        /**
+         * build the routing ProduceRequestValidator
+         * @return validator
+         */
+        public ProduceRequestValidator build() {
+            return new RoutingProduceRequestValidator(routingRules, defaultValidator);
+        }
+
+    }
+}
diff --git a/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/topic/AllValidTopicValidationResult.java b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/topic/AllValidTopicValidationResult.java
new file mode 100644
index 00000000..b50ea271
--- /dev/null
+++ b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/topic/AllValidTopicValidationResult.java
@@ -0,0 +1,33 @@
+/*
+ * 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.filter.schema.validation.topic;
+
+import java.util.List;
+import java.util.stream.Stream;
+
+record AllValidTopicValidationResult(String topicName) implements TopicValidationResult {
+
+    @Override
+    public boolean isAnyPartitionInvalid() {
+        return false;
+    }
+
+    @Override
+    public boolean isAllPartitionsInvalid() {
+        return false;
+    }
+
+    @Override
+    public Stream<PartitionValidationResult> invalidPartitions() {
+        return Stream.empty();
+    }
+
+    @Override
+    public PartitionValidationResult getPartitionResult(int index) {
+        return new PartitionValidationResult(index, List.of());
+    }
+}
diff --git a/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/topic/AllValidTopicValidator.java b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/topic/AllValidTopicValidator.java
new file mode 100644
index 00000000..9da9c08b
--- /dev/null
+++ b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/topic/AllValidTopicValidator.java
@@ -0,0 +1,16 @@
+/*
+ * 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.filter.schema.validation.topic;
+
+import org.apache.kafka.common.message.ProduceRequestData;
+
+class AllValidTopicValidator implements TopicValidator {
+    @Override
+    public TopicValidationResult validateTopicData(ProduceRequestData.TopicProduceData request) {
+        return new AllValidTopicValidationResult(request.name());
+    }
+}
diff --git a/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/topic/PartitionValidationResult.java b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/topic/PartitionValidationResult.java
new file mode 100644
index 00000000..649ee1ee
--- /dev/null
+++ b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/topic/PartitionValidationResult.java
@@ -0,0 +1,25 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.proxy.filter.schema.validation.topic;
+
+import java.util.List;
+
+/**
+ * Result of the validation of a single partition from a ProduceRequestData
+ * @param index index of the partition
+ * @param recordValidationFailures details of any records that failed validation
+ */
+public record PartitionValidationResult(int index, List<RecordValidationFailure> recordValidationFailures) {
+
+    /**
+     * Are all records in the partition valid?
+     * @return true if all records in the partition are valid
+     */
+    public boolean allRecordsValid() {
+        return recordValidationFailures.isEmpty();
+    }
+}
diff --git a/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/topic/PerPartitionTopicValidationResult.java b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/topic/PerPartitionTopicValidationResult.java
new file mode 100644
index 00000000..dfa6ca6d
--- /dev/null
+++ b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/topic/PerPartitionTopicValidationResult.java
@@ -0,0 +1,33 @@
+/*
+ * 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.filter.schema.validation.topic;
+
+import java.util.Map;
+import java.util.stream.Stream;
+
+record PerPartitionTopicValidationResult(String topicName, Map<Integer, PartitionValidationResult> partitionValidationResults) implements TopicValidationResult {
+
+    @Override
+    public boolean isAnyPartitionInvalid() {
+        return partitionValidationResults.values().stream().anyMatch(partitionValidationResult -> !partitionValidationResult.allRecordsValid());
+    }
+
+    @Override
+    public boolean isAllPartitionsInvalid() {
+        return partitionValidationResults.values().stream().noneMatch(PartitionValidationResult::allRecordsValid);
+    }
+
+    @Override
+    public Stream<PartitionValidationResult> invalidPartitions() {
+        return partitionValidationResults.values().stream().filter(partitionValidationResult -> !partitionValidationResult.allRecordsValid());
+    }
+
+    @Override
+    public PartitionValidationResult getPartitionResult(int index) {
+        return partitionValidationResults.get(index);
+    }
+}
diff --git a/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/topic/PerRecordTopicValidator.java b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/topic/PerRecordTopicValidator.java
new file mode 100644
index 00000000..adf55634
--- /dev/null
+++ b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/topic/PerRecordTopicValidator.java
@@ -0,0 +1,60 @@
+/*
+ * 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.filter.schema.validation.topic;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import org.apache.kafka.common.message.ProduceRequestData;
+import org.apache.kafka.common.record.BaseRecords;
+import org.apache.kafka.common.record.MemoryRecords;
+import org.apache.kafka.common.record.MutableRecordBatch;
+import org.apache.kafka.common.record.Record;
+
+import io.kroxylicious.proxy.filter.schema.validation.Result;
+import io.kroxylicious.proxy.filter.schema.validation.record.RecordValidator;
+
+class PerRecordTopicValidator implements TopicValidator {
+
+    private final RecordValidator validator;
+
+    PerRecordTopicValidator(RecordValidator validator) {
+        if (validator == null) {
+            throw new IllegalArgumentException("validator was null");
+        }
+        this.validator = validator;
+    }
+
+    @Override
+    public TopicValidationResult validateTopicData(ProduceRequestData.TopicProduceData topicProduceData) {
+        return new PerPartitionTopicValidationResult(topicProduceData.name(), topicProduceData.partitionData().stream().collect(Collectors.toMap(
+                ProduceRequestData.PartitionProduceData::index, this::validateTopicPartition)));
+    }
+
+    private PartitionValidationResult validateTopicPartition(ProduceRequestData.PartitionProduceData partitionProduceData) {
+        return new PartitionValidationResult(partitionProduceData.index(), validateRecords(partitionProduceData.records()));
+    }
+
+    private List<RecordValidationFailure> validateRecords(BaseRecords records) {
+        if (!(records instanceof MemoryRecords)) {
+            return List.of();
+        }
+        int recordIndex = 0;
+        List<RecordValidationFailure> failures = new ArrayList<>();
+        for (MutableRecordBatch batch : ((MemoryRecords) records).batches()) {
+            for (Record record : batch) {
+                Result result = validator.validate(record);
+                if (!result.valid()) {
+                    failures.add(new RecordValidationFailure(recordIndex, result.errorMessage()));
+                }
+                recordIndex++;
+            }
+        }
+        return failures;
+    }
+}
diff --git a/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/topic/RecordValidationFailure.java b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/topic/RecordValidationFailure.java
new file mode 100644
index 00000000..1efb9257
--- /dev/null
+++ b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/topic/RecordValidationFailure.java
@@ -0,0 +1,20 @@
+/*
+ * 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.filter.schema.validation.topic;
+
+/**
+ * Details about an invalid record from a topic-partition
+ * @param invalidIndex the index of the invalid record within it's batch
+ * @param errorMessage details of what was invalid
+ */
+public record RecordValidationFailure(int invalidIndex, String errorMessage) {
+
+    @Override
+    public String toString() {
+        return "(" + invalidIndex + ", " + errorMessage + ")";
+    }
+}
diff --git a/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/topic/TopicValidationResult.java b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/topic/TopicValidationResult.java
new file mode 100644
index 00000000..a45d6b8e
--- /dev/null
+++ b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/topic/TopicValidationResult.java
@@ -0,0 +1,46 @@
+/*
+ * 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.filter.schema.validation.topic;
+
+import java.util.stream.Stream;
+
+/**
+ * Result of validating TopicProducerData
+ */
+public interface TopicValidationResult {
+
+    /**
+     * is any partition invalid
+     * @return true if any partition invalid
+     */
+    boolean isAnyPartitionInvalid();
+
+    /**
+     * are all partitions invalid
+     * @return true if all partitions invalid
+     */
+    boolean isAllPartitionsInvalid();
+
+    /**
+     * get invalid partitions
+     * @return stream of invalid partitions
+     */
+    Stream<PartitionValidationResult> invalidPartitions();
+
+    /**
+     * name of validated topic
+     * @return topicName
+     */
+    String topicName();
+
+    /**
+     * get partition result
+     * @param index partition index
+     * @return result for partition
+     */
+    PartitionValidationResult getPartitionResult(int index);
+}
diff --git a/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/topic/TopicValidator.java b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/topic/TopicValidator.java
new file mode 100644
index 00000000..4fbce1d2
--- /dev/null
+++ b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/topic/TopicValidator.java
@@ -0,0 +1,22 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.proxy.filter.schema.validation.topic;
+
+import org.apache.kafka.common.message.ProduceRequestData;
+
+/**
+ * Validates ${@link org.apache.kafka.common.message.ProduceRequestData.TopicProduceData}
+ */
+public interface TopicValidator {
+    /**
+     * Validate topic produce data, returning details about which partitions/records were
+     * invalid
+     * @param request the request
+     * @return result describing whether any partitions were invalid, and details of any invalid partitions/records
+     */
+    TopicValidationResult validateTopicData(ProduceRequestData.TopicProduceData request);
+}
diff --git a/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/topic/TopicValidators.java b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/topic/TopicValidators.java
new file mode 100644
index 00000000..3ea4d45c
--- /dev/null
+++ b/kroxylicious-schema-validation/src/main/java/io/kroxylicious/proxy/filter/schema/validation/topic/TopicValidators.java
@@ -0,0 +1,38 @@
+/*
+ * 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.filter.schema.validation.topic;
+
+import io.kroxylicious.proxy.filter.schema.validation.record.RecordValidator;
+
+/**
+ * Factory for different topic ${@link TopicValidator} implementations
+ */
+public class TopicValidators {
+
+    private TopicValidators() {
+    }
+
+    private static final AllValidTopicValidator ALL_VALID = new AllValidTopicValidator();
+
+    /**
+     * A validator that always validates any input topic data
+     * @return validator
+     */
+    public static TopicValidator allValid() {
+        return ALL_VALID;
+    }
+
+    /**
+     * A validator that tests the records of a topic produce data and invalidates if any records are invalid
+     * @param validator a validator
+     * @return per-record topic validator
+     */
+    public static TopicValidator perRecordValidator(RecordValidator validator) {
+        return new PerRecordTopicValidator(validator);
+    }
+
+}
diff --git a/kroxylicious-schema-validation/src/main/resources/META-INF/services/io.kroxylicious.proxy.filter.FilterContributor b/kroxylicious-schema-validation/src/main/resources/META-INF/services/io.kroxylicious.proxy.filter.FilterContributor
new file mode 100644
index 00000000..57e8566c
--- /dev/null
+++ b/kroxylicious-schema-validation/src/main/resources/META-INF/services/io.kroxylicious.proxy.filter.FilterContributor
@@ -0,0 +1,1 @@
+io.kroxylicious.proxy.filter.schema.ProduceRequestValidationFilterContributor
diff --git a/pom.xml b/pom.xml
index 19ff6a54..eaac67d9 100644
--- a/pom.xml
+++ b/pom.xml
@@ -218,6 +218,7 @@
     <modules>
         <module>api/kroxylicious-api</module>
         <module>api/kroxylicious-filter-api</module>
+        <module>kroxylicious-schema-validation</module>
         <module>krpc-code-gen</module>
         <module>kroxylicious-test-tools</module>
         <module>kroxylicious</module>


Test patch:
diff --git a/integrationtests/pom.xml b/integrationtests/pom.xml
index c6ba825f..36ee2698 100644
--- a/integrationtests/pom.xml
+++ b/integrationtests/pom.xml
@@ -43,6 +43,12 @@
             <version>0.2.0-SNAPSHOT</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>io.kroxylicious</groupId>
+            <artifactId>kroxylicious-schema-validation</artifactId>
+            <version>0.2.0-SNAPSHOT</version>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>io.kroxylicious.testing</groupId>
             <artifactId>testing-junit5-extension</artifactId>
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/schema/validation/JsonSyntaxValidationIT.java b/integrationtests/src/test/java/io/kroxylicious/proxy/schema/validation/JsonSyntaxValidationIT.java
new file mode 100644
index 00000000..ac4d2ecd
--- /dev/null
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/schema/validation/JsonSyntaxValidationIT.java
@@ -0,0 +1,298 @@
+/*
+ * 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.schema.validation;
+
+import java.time.Duration;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
+import org.apache.kafka.common.InvalidRecordException;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.assertj.core.api.Assertions;
+import org.jetbrains.annotations.NotNull;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import io.kroxylicious.proxy.KroxyConfig;
+import io.kroxylicious.proxy.KroxyConfigBuilder;
+import io.kroxylicious.proxy.VirtualClusterBuilder;
+import io.kroxylicious.testing.kafka.api.KafkaCluster;
+import io.kroxylicious.testing.kafka.junit5ext.KafkaClusterExtension;
+
+import static io.kroxylicious.proxy.Utils.startProxy;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+@ExtendWith(KafkaClusterExtension.class)
+public class JsonSyntaxValidationIT {
+
+    public static final String SYNTACTICALLY_CORRECT_JSON = "{\"value\":\"json\"}";
+    public static final String SYNTACTICALLY_INCORRECT_JSON = "Not Json";
+    private static final String TOPIC_1 = "my-test-topic";
+    private static final String TOPIC_2 = "my-test-topic-2";
+
+    @Test
+    public void testInvalidJsonProduceRejected(KafkaCluster cluster, Admin admin) throws Exception {
+        String proxyAddress = "localhost:9192";
+
+        admin.createTopics(List.of(new NewTopic(TOPIC_1, 1, (short) 1))).all().get();
+
+        var config = baseConfigBuilder(cluster, proxyAddress).addNewFilter().withType("ProduceValidator").withConfig(Map.of("rules",
+                List.of(Map.of("topicNames", List.of(TOPIC_1), "valueRule",
+                        Map.of("allowsNulls", true, "syntacticallyCorrectJson", Map.of("validateObjectKeysUnique", true))))))
+                .endFilter()
+                .build().toYaml();
+
+        try (var ignored = startProxy(config)) {
+            try (var producer = getProducer(proxyAddress, 0, 16384)) {
+                Future<RecordMetadata> invalid = producer.send(new ProducerRecord<>(TOPIC_1, "my-key", SYNTACTICALLY_INCORRECT_JSON));
+                assertInvalidRecordExceptionThrown(invalid, "value was not syntactically correct JSON");
+            }
+        }
+    }
+
+    private static KroxyConfigBuilder baseConfigBuilder(KafkaCluster cluster, String proxyAddress) {
+        assertEquals(1, cluster.getNumOfBrokers());
+        String bootstrapServers = cluster.getBootstrapServers();
+        return KroxyConfig.builder().addToVirtualClusters("demo",
+                new VirtualClusterBuilder().withNewTargetCluster().withBootstrapServers(bootstrapServers).endTargetCluster().withNewClusterEndpointConfigProvider()
+                        .withType("StaticCluster").withConfig(Map.of("bootstrapAddress", proxyAddress)).endClusterEndpointConfigProvider().build())
+                .addNewFilter()
+                .withType("ApiVersions").endFilter().addNewFilter().withType("BrokerAddress").endFilter();
+    }
+
+    @Test
+    public void testInvalidJsonProduceRejectedUsingTopicNames(KafkaCluster cluster, Admin admin) throws Exception {
+        String proxyAddress = "localhost:9192";
+
+        admin.createTopics(List.of(new NewTopic(TOPIC_1, 1, (short) 1), new NewTopic(TOPIC_2, 1, (short) 1))).all().get();
+
+        var config = baseConfigBuilder(cluster, proxyAddress).addNewFilter().withType("ProduceValidator").withConfig(Map.of("rules",
+                List.of(Map.of("topicNames", List.of(TOPIC_1), "valueRule",
+                        Map.of("allowsNulls", true, "syntacticallyCorrectJson", Map.of("validateObjectKeysUnique", true))))))
+                .endFilter()
+                .build().toYaml();
+
+        try (var ignored = startProxy(config)) {
+            try (var producer = getProducer(proxyAddress, 0, 16384)) {
+                Future<RecordMetadata> invalid = producer.send(new ProducerRecord<>(TOPIC_1, "my-key", SYNTACTICALLY_INCORRECT_JSON));
+                assertInvalidRecordExceptionThrown(invalid, "value was not syntactically correct JSON");
+
+                producer.send(new ProducerRecord<>(TOPIC_2, "my-key", SYNTACTICALLY_INCORRECT_JSON)).get();
+                try (var consumer = new KafkaConsumer<String, String>(
+                        Map.of(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, proxyAddress, ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class,
+                                ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class, ConsumerConfig.GROUP_ID_CONFIG, "my-group-id",
+                                ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"))) {
+                    consumer.subscribe(Set.of(TOPIC_2));
+                    var records = consumer.poll(Duration.ofSeconds(10));
+                    assertEquals(1, records.count());
+                    assertEquals(SYNTACTICALLY_INCORRECT_JSON, records.iterator().next().value());
+                }
+            }
+        }
+    }
+
+    @Test
+    public void testPartiallyInvalidJsonTransactionalAllRejected(KafkaCluster cluster, Admin admin) throws Exception {
+        String proxyAddress = "localhost:9192";
+
+        admin.createTopics(List.of(new NewTopic(TOPIC_1, 1, (short) 1), new NewTopic(TOPIC_2, 1, (short) 1))).all().get();
+
+        var config = baseConfigBuilder(cluster, proxyAddress).addNewFilter().withType("ProduceValidator").withConfig(
+                Map.of("forwardPartialRequests", true, "rules", List.of(Map.of("topicNames", List.of(TOPIC_1, TOPIC_2), "valueRule",
+                        Map.of("allowsNulls", true, "syntacticallyCorrectJson", Map.of("validateObjectKeysUnique", true))))))
+                .endFilter()
+                .build().toYaml();
+
+        try (var ignored = startProxy(config)) {
+            try (var producer = getProducer(proxyAddress,
+                    Map.of(ProducerConfig.LINGER_MS_CONFIG, 5000, ProducerConfig.TRANSACTIONAL_ID_CONFIG, UUID.randomUUID().toString()))) {
+                producer.initTransactions();
+                producer.beginTransaction();
+                Future<RecordMetadata> invalid = producer.send(new ProducerRecord<>(TOPIC_1, "my-key", SYNTACTICALLY_INCORRECT_JSON));
+                Future<RecordMetadata> valid = producer.send(new ProducerRecord<>(TOPIC_2, "my-key", SYNTACTICALLY_CORRECT_JSON));
+                producer.flush();
+                assertInvalidRecordExceptionThrown(invalid, "value was not syntactically correct JSON");
+                assertInvalidRecordExceptionThrown(valid, "Invalid record in another topic-partition caused whole ProduceRequest to be invalidated");
+                producer.abortTransaction();
+            }
+        }
+    }
+
+    @Test
+    public void testPartiallyInvalidJsonNotConfiguredToForwardAllRejected(KafkaCluster cluster, Admin admin) throws Exception {
+        String proxyAddress = "localhost:9192";
+
+        admin.createTopics(List.of(new NewTopic(TOPIC_1, 1, (short) 1), new NewTopic(TOPIC_2, 1, (short) 1))).all().get();
+
+        boolean forwardPartialRequests = false;
+        var config = baseConfigBuilder(cluster, proxyAddress).addNewFilter().withType("ProduceValidator").withConfig(
+                Map.of("forwardPartialRequests", forwardPartialRequests, "rules", List.of(Map.of("topicNames", List.of(TOPIC_1, TOPIC_2), "valueRule",
+                        Map.of("allowsNulls", true, "syntacticallyCorrectJson", Map.of("validateObjectKeysUnique", true))))))
+                .endFilter()
+                .build().toYaml();
+
+        try (var ignored = startProxy(config)) {
+            try (var producer = getProducer(proxyAddress, 5000, 16384)) {
+                Future<RecordMetadata> invalid = producer.send(new ProducerRecord<>(TOPIC_1, "my-key", SYNTACTICALLY_INCORRECT_JSON));
+                Future<RecordMetadata> valid = producer.send(new ProducerRecord<>(TOPIC_2, "my-key", SYNTACTICALLY_CORRECT_JSON));
+                producer.flush();
+                assertInvalidRecordExceptionThrown(invalid, "value was not syntactically correct JSON");
+                assertInvalidRecordExceptionThrown(valid, "Invalid record in another topic-partition caused whole ProduceRequest to be invalidated");
+            }
+        }
+    }
+
+    @Test
+    public void testPartiallyInvalidJsonProduceRejected(KafkaCluster cluster, Admin admin) throws Exception {
+        String proxyAddress = "localhost:9192";
+
+        admin.createTopics(List.of(new NewTopic(TOPIC_1, 1, (short) 1), new NewTopic(TOPIC_2, 1, (short) 1))).all().get();
+
+        var config = baseConfigBuilder(cluster, proxyAddress).addNewFilter().withType("ProduceValidator").withConfig(
+                Map.of("forwardPartialRequests", true, "rules", List.of(Map.of("topicNames", List.of(TOPIC_1, TOPIC_2), "valueRule",
+                        Map.of("allowsNulls", true, "syntacticallyCorrectJson", Map.of("validateObjectKeysUnique", true))))))
+                .endFilter()
+                .build().toYaml();
+
+        try (var ignored = startProxy(config)) {
+            try (var producer = getProducer(proxyAddress, 5000, 16384)) {
+                Future<RecordMetadata> invalid = producer.send(new ProducerRecord<>(TOPIC_1, "my-key", SYNTACTICALLY_INCORRECT_JSON));
+                Future<RecordMetadata> valid = producer.send(new ProducerRecord<>(TOPIC_2, "my-key", SYNTACTICALLY_CORRECT_JSON));
+                producer.flush();
+                assertInvalidRecordExceptionThrown(invalid, "value was not syntactically correct JSON");
+                RecordMetadata metadata = valid.get(10, TimeUnit.SECONDS);
+                assertTrue(metadata.hasOffset());
+            }
+
+            try (var consumer = new KafkaConsumer<String, String>(
+                    Map.of(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, proxyAddress, ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class,
+                            ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class, ConsumerConfig.GROUP_ID_CONFIG, "my-group-id",
+                            ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"))) {
+                consumer.subscribe(Set.of(TOPIC_2));
+                var records = consumer.poll(Duration.ofSeconds(10));
+                assertEquals(1, records.count());
+                assertEquals(SYNTACTICALLY_CORRECT_JSON, records.iterator().next().value());
+            }
+        }
+    }
+
+    @Test
+    public void testPartiallyInvalidAcrossPartitionsOfSameTopic(KafkaCluster cluster, Admin admin) throws Exception {
+        String proxyAddress = "localhost:9192";
+
+        admin.createTopics(List.of(new NewTopic(TOPIC_1, 2, (short) 1))).all().get();
+
+        var config = baseConfigBuilder(cluster, proxyAddress).addNewFilter().withType("ProduceValidator").withConfig(
+                Map.of("forwardPartialRequests", true, "rules", List.of(Map.of("topicNames", List.of(TOPIC_1), "valueRule",
+                        Map.of("allowsNulls", true, "syntacticallyCorrectJson", Map.of("validateObjectKeysUnique", true))))))
+                .endFilter()
+                .build().toYaml();
+
+        try (var ignored = startProxy(config)) {
+            try (var producer = getProducer(proxyAddress, 5000, 16384)) {
+                Future<RecordMetadata> invalid = producer.send(new ProducerRecord<>(TOPIC_1, 0, "my-key", SYNTACTICALLY_INCORRECT_JSON));
+                Future<RecordMetadata> valid = producer.send(new ProducerRecord<>(TOPIC_1, 1, "my-key", SYNTACTICALLY_CORRECT_JSON));
+                producer.flush();
+                assertInvalidRecordExceptionThrown(invalid, "value was not syntactically correct JSON");
+                RecordMetadata metadata = valid.get(10, TimeUnit.SECONDS);
+                assertTrue(metadata.hasOffset());
+            }
+
+            try (var consumer = new KafkaConsumer<String, String>(
+                    Map.of(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, proxyAddress, ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class,
+                            ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class, ConsumerConfig.GROUP_ID_CONFIG, "my-group-id",
+                            ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"))) {
+                consumer.subscribe(Set.of(TOPIC_1));
+                var records = consumer.poll(Duration.ofSeconds(10));
+                assertEquals(1, records.count());
+                assertEquals(SYNTACTICALLY_CORRECT_JSON, records.iterator().next().value());
+            }
+        }
+    }
+
+    @Test
+    public void testPartiallyInvalidWithinOnePartitionOfTopic(KafkaCluster cluster, Admin admin) throws Exception {
+        String proxyAddress = "localhost:9192";
+
+        admin.createTopics(List.of(new NewTopic(TOPIC_1, 1, (short) 1))).all().get();
+
+        var config = baseConfigBuilder(cluster, proxyAddress).addNewFilter().withType("ProduceValidator").withConfig(
+                Map.of("forwardPartialRequests", true, "rules", List.of(Map.of("topicNames", List.of(TOPIC_1), "valueRule",
+                        Map.of("allowsNulls", true, "syntacticallyCorrectJson", Map.of("validateObjectKeysUnique", true))))))
+                .endFilter()
+                .build().toYaml();
+
+        try (var ignored = startProxy(config)) {
+            try (var producer = getProducer(proxyAddress, 5000, 16384)) {
+                Future<RecordMetadata> invalid = producer.send(new ProducerRecord<>(TOPIC_1, "my-key", SYNTACTICALLY_INCORRECT_JSON));
+                Future<RecordMetadata> valid = producer.send(new ProducerRecord<>(TOPIC_1, "my-key", SYNTACTICALLY_CORRECT_JSON));
+                producer.flush();
+                assertInvalidRecordExceptionThrown(invalid, "value was not syntactically correct JSON");
+                Assertions.assertThatThrownBy(() -> {
+                    valid.get(10, TimeUnit.SECONDS);
+                }).isInstanceOf(ExecutionException.class).hasCauseInstanceOf(KafkaException.class).cause()
+                        .hasMessageContaining("Failed to append record because it was part of a batch which had one more more invalid records");
+            }
+        }
+    }
+
+    @Test
+    public void testValidJsonProduceAccepted(KafkaCluster cluster, Admin admin) throws Exception {
+        String proxyAddress = "localhost:9192";
+
+        admin.createTopics(List.of(new NewTopic(TOPIC_1, 1, (short) 1))).all().get();
+
+        var config = baseConfigBuilder(cluster, proxyAddress).addNewFilter().withType("ProduceValidator").withConfig(Map.of("rules",
+                List.of(Map.of("topicNames", List.of(TOPIC_1), "valueRule",
+                        Map.of("allowsNulls", true, "syntacticallyCorrectJson", Map.of("validateObjectKeysUnique", true))))))
+                .endFilter()
+                .build().toYaml();
+
+        try (var ignored = startProxy(config)) {
+            try (var producer = getProducer(proxyAddress, 0, 16384)) {
+                producer.send(new ProducerRecord<>(TOPIC_1, "my-key", SYNTACTICALLY_CORRECT_JSON)).get();
+            }
+        }
+    }
+
+    @NotNull
+    private static KafkaProducer<String, String> getProducer(String proxyAddress, int linger, int batchSize) {
+        return getProducer(proxyAddress, Map.of(ProducerConfig.LINGER_MS_CONFIG, linger, ProducerConfig.BATCH_SIZE_CONFIG, batchSize));
+    }
+
+    private static KafkaProducer<String, String> getProducer(String proxyAddress, Map<String, Object> additionalProps) {
+        Map<String, Object> produceProps = new java.util.HashMap<>(Map.of(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, proxyAddress, ProducerConfig.CLIENT_ID_CONFIG,
+                "shouldModifyProduceMessage",
+                ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class,
+                ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class, ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 3_600_000));
+        produceProps.putAll(additionalProps);
+        return new KafkaProducer<>(produceProps);
+    }
+
+    private static void assertInvalidRecordExceptionThrown(Future<RecordMetadata> invalid, String message) {
+        Assertions.assertThatThrownBy(() -> {
+            invalid.get(10, TimeUnit.SECONDS);
+        }).isInstanceOf(ExecutionException.class).hasCauseInstanceOf(InvalidRecordException.class).cause()
+                .hasMessageContaining(message);
+    }
+
+}
diff --git a/kroxylicious-schema-validation/src/test/java/io/kroxylicious/proxy/filter/schema/JsonSyntaxBytebufValidatorTest.java b/kroxylicious-schema-validation/src/test/java/io/kroxylicious/proxy/filter/schema/JsonSyntaxBytebufValidatorTest.java
new file mode 100644
index 00000000..f91f6799
--- /dev/null
+++ b/kroxylicious-schema-validation/src/test/java/io/kroxylicious/proxy/filter/schema/JsonSyntaxBytebufValidatorTest.java
@@ -0,0 +1,224 @@
+/*
+ * 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.filter.schema;
+
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+
+import org.apache.kafka.common.record.DefaultRecord;
+import org.apache.kafka.common.record.Record;
+import org.apache.kafka.common.utils.ByteBufferOutputStream;
+import org.junit.jupiter.api.Test;
+
+import io.kroxylicious.proxy.filter.schema.validation.Result;
+import io.kroxylicious.proxy.filter.schema.validation.bytebuf.BytebufValidator;
+import io.kroxylicious.proxy.filter.schema.validation.bytebuf.BytebufValidators;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class JsonSyntaxBytebufValidatorTest {
+
+    @Test
+    public void testSyntacticallyIncorrectRecordInvalidated() {
+        Record record = createRecord("a", "b");
+        BytebufValidator validator = BytebufValidators.jsonSyntaxValidator(false);
+        Result result = validate(record, validator);
+        assertFalse(result.valid());
+    }
+
+    @Test
+    public void testSyntacticallyCorrectRecordValidated() {
+        Record record = createRecord("a", "{\"a\":\"a\"}");
+        BytebufValidator validator = BytebufValidators.jsonSyntaxValidator(false);
+        Result result = validate(record, validator);
+        assertTrue(result.valid());
+    }
+
+    @Test
+    public void testDuplicatedObjectKeyInvalidated() {
+        Record record = createRecord("a", "{\"a\":\"a\",\"a\":\"b\"}");
+        BytebufValidator validator = BytebufValidators.jsonSyntaxValidator(true);
+        Result result = validate(record, validator);
+        assertFalse(result.valid());
+        assertTrue(result.errorMessage().contains("value was not syntactically correct JSON: Duplicate field"));
+    }
+
+    @Test
+    public void testDuplicatedObjectKeyInNestedObjectInvalidated() {
+        Record record = createRecord("a", "{\"inner\":{\"a\":\"a\",\"a\":\"b\"}}");
+        BytebufValidator validator = BytebufValidators.jsonSyntaxValidator(true);
+        Result result = validate(record, validator);
+        assertFalse(result.valid());
+        assertTrue(result.errorMessage().contains("value was not syntactically correct JSON: Duplicate field"));
+    }
+
+    @Test
+    public void testDuplicatedObjectKeyInArrayInvalidated() {
+        Record record = createRecord("a", "[{\"a\":\"a\",\"a\":\"b\"}]");
+        BytebufValidator validator = BytebufValidators.jsonSyntaxValidator(true);
+        Result result = validate(record, validator);
+        assertFalse(result.valid());
+        assertTrue(result.errorMessage().contains("value was not syntactically correct JSON: Duplicate field"));
+    }
+
+    @Test
+    public void testNonDuplicatedObjectKeyInArrayValidated() {
+        Record record = createRecord("a", "[{\"a\":\"a\",\"b\":\"b\"}]");
+        BytebufValidator validator = BytebufValidators.jsonSyntaxValidator(true);
+        Result result = validate(record, validator);
+        assertTrue(result.valid());
+    }
+
+    @Test
+    public void testArrayWithTwoObjectsWithSameKeysValidated() {
+        Record record = createRecord("a", "[{\"a\":\"a\"},{\"a\":\"a\"}]");
+        BytebufValidator validator = BytebufValidators.jsonSyntaxValidator(true);
+        Result result = validate(record, validator);
+        assertTrue(result.valid());
+    }
+
+    @Test
+    public void testNestedObjectsUsingSameKeysValidated() {
+        Record record = createRecord("a", "[{\"a\":{\"a\":\"a\"}}]");
+        BytebufValidator validator = BytebufValidators.jsonSyntaxValidator(true);
+        Result result = validate(record, validator);
+        assertTrue(result.valid());
+    }
+
+    @Test
+    public void testNestedObjectsWithDuplicateKeysInvalidated() {
+        Record record = createRecord("a", "[{\"a\":{\"a\":\"a\",\"a\":\"b\"}}]");
+        BytebufValidator validator = BytebufValidators.jsonSyntaxValidator(true);
+        Result result = validate(record, validator);
+        assertFalse(result.valid());
+        assertTrue(result.errorMessage().contains("value was not syntactically correct JSON: Duplicate field"));
+    }
+
+    @Test
+    public void testDeepObjectsWithDuplicateKeysInvalidated() {
+        Record record = createRecord("a", "[[[{\"a\":{\"b\":[1,true,null,{\"duplicate\":1,\"duplicate\":1}]}}]]]]");
+        BytebufValidator validator = BytebufValidators.jsonSyntaxValidator(true);
+        Result result = validate(record, validator);
+        assertFalse(result.valid());
+        assertTrue(result.errorMessage().contains("value was not syntactically correct JSON: Duplicate field"));
+    }
+
+    @Test
+    public void testArrayWithTwoObjectsWithSameKeysAndOtherDataValidated() {
+        Record record = createRecord("a", "[{\"a\":\"a\"},2,{\"a\":\"a\"},\"banana\"]");
+        BytebufValidator validator = BytebufValidators.jsonSyntaxValidator(true);
+        Result result = validate(record, validator);
+        assertTrue(result.valid());
+    }
+
+    @Test
+    public void testNonDuplicatedObjectKeysWithDuplicationValidationEnabled() {
+        Record record = createRecord("a", "{\"a\":\"b\",\"c\":\"d\"}");
+        BytebufValidator validator = BytebufValidators.jsonSyntaxValidator(true);
+        Result result = validate(record, validator);
+        assertTrue(result.valid());
+    }
+
+    @Test
+    public void testDuplicatedObjectKeyValidatedWithDuplicationValidationDisabled() {
+        Record record = createRecord("a", "{\"a\":\"a\",\"a\":\"b\"}");
+        BytebufValidator validator = BytebufValidators.jsonSyntaxValidator(false);
+        Result result = validate(record, validator);
+        assertTrue(result.valid());
+    }
+
+    @Test
+    public void testDifferentObjectsCanHaveSameKeyNames() {
+        Record record = createRecord("a", "{\"a\":{\"a\":1},\"b\":{\"a\":2}}");
+        BytebufValidator validator = BytebufValidators.jsonSyntaxValidator(true);
+        Result result = validate(record, validator);
+        assertTrue(result.valid());
+    }
+
+    @Test
+    public void testTrailingCharactersInvalidated() {
+        Record record = createRecord("a", "{\"a\":\"a\"}abc");
+        BytebufValidator validator = BytebufValidators.jsonSyntaxValidator(false);
+        Result result = validate(record, validator);
+        assertFalse(result.valid());
+    }
+
+    @Test
+    public void testLeadingCharactersInvalidated() {
+        Record record = createRecord("a", "abc{\"a\":\"a\"}abc");
+        BytebufValidator validator = BytebufValidators.jsonSyntaxValidator(false);
+        Result result = validate(record, validator);
+        assertFalse(result.valid());
+    }
+
+    @Test
+    public void testValueValidated() {
+        Record record = createRecord("a", "123");
+        BytebufValidator validator = BytebufValidators.jsonSyntaxValidator(true);
+        Result result = validate(record, validator);
+        assertTrue(result.valid());
+    }
+
+    @Test
+    public void testKeyValidated() {
+        Record record = createRecord("\"abc\"", "123");
+        BytebufValidator validator = BytebufValidators.jsonSyntaxValidator(true);
+        Result result = validator.validate(record.key(), record.keySize(), record, true);
+        assertTrue(result.valid());
+    }
+
+    @Test
+    public void testEmptyStringThrows() {
+        Record record = createRecord("a", "");
+        BytebufValidator validator = BytebufValidators.jsonSyntaxValidator(false);
+        assertThrows(IllegalArgumentException.class, () -> {
+            validate(record, validator);
+        });
+    }
+
+    @Test
+    public void testNullValueThrows() {
+        Record record = createRecord("a", null);
+        BytebufValidator validator = BytebufValidators.jsonSyntaxValidator(false);
+        assertThrows(IllegalArgumentException.class, () -> {
+            validate(record, validator);
+        });
+    }
+
+    private static Result validate(Record record, BytebufValidator validator) {
+        return validator.validate(record.value(), record.valueSize(), record, false);
+    }
+
+    private Record createRecord(String key, String value) {
+        ByteBuffer keyBuf = toBufNullable(key);
+        ByteBuffer valueBuf = toBufNullable(value);
+
+        try (ByteBufferOutputStream bufferOutputStream = new ByteBufferOutputStream(1000); DataOutputStream dataOutputStream = new DataOutputStream(bufferOutputStream)) {
+            DefaultRecord.writeTo(dataOutputStream, 0, 0, keyBuf, valueBuf, Record.EMPTY_HEADERS);
+            dataOutputStream.flush();
+            bufferOutputStream.flush();
+            ByteBuffer buffer = bufferOutputStream.buffer();
+            buffer.flip();
+            return DefaultRecord.readFrom(buffer, 0, 0, 0, 0L);
+        }
+        catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    private static ByteBuffer toBufNullable(String key) {
+        if (key == null) {
+            return null;
+        }
+        byte[] keyBytes = key.getBytes(StandardCharsets.UTF_8);
+        return ByteBuffer.wrap(keyBytes);
+    }
+}
diff --git a/kroxylicious-schema-validation/src/test/java/io/kroxylicious/proxy/filter/schema/config/ValidationConfigTest.java b/kroxylicious-schema-validation/src/test/java/io/kroxylicious/proxy/filter/schema/config/ValidationConfigTest.java
new file mode 100644
index 00000000..01627105
--- /dev/null
+++ b/kroxylicious-schema-validation/src/test/java/io/kroxylicious/proxy/filter/schema/config/ValidationConfigTest.java
@@ -0,0 +1,76 @@
+/*
+ * 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.filter.schema.config;
+
+import java.util.List;
+import java.util.Set;
+
+import org.junit.jupiter.api.Test;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+class ValidationConfigTest {
+
+    @Test
+    public void testDecodeDefaultValues() throws JsonProcessingException {
+        ObjectMapper yamlMapper = new ObjectMapper(new YAMLFactory());
+        ValidationConfig deserialised = yamlMapper.readerFor(ValidationConfig.class).readValue("""
+                defaultRule:
+                  valueRule: {}
+                rules:
+                - topicNames:
+                  - one
+                  valueRule:
+                    syntacticallyCorrectJson: {}
+                - topicNames:
+                  - two
+                  keyRule: {}
+                """);
+
+        TopicMatchingRecordValidationRule ruleOne = new TopicMatchingRecordValidationRule(Set.of("one"), null,
+                new BytebufValidation(new SyntacticallyCorrectJsonConfig(false), true, false));
+        TopicMatchingRecordValidationRule ruleTwo = new TopicMatchingRecordValidationRule(Set.of("two"), new BytebufValidation(null, true, false), null);
+        ValidationConfig expected = new ValidationConfig(false, List.of(ruleOne, ruleTwo), new RecordValidationRule(null, new BytebufValidation(null, true, false)));
+        assertEquals(expected, deserialised);
+    }
+
+    @Test
+    public void testDecodeNonDefaultValues() throws JsonProcessingException {
+        ObjectMapper yamlMapper = new ObjectMapper(new YAMLFactory());
+        ValidationConfig deserialised = yamlMapper.readerFor(ValidationConfig.class).readValue("""
+                forwardPartialRequests: true
+                defaultRule:
+                  valueRule:
+                    allowNulls: false
+                    allowEmpty: true
+                rules:
+                - topicNames:
+                  - one
+                  valueRule:
+                    syntacticallyCorrectJson:
+                        validateObjectKeysUnique: true
+                    allowNulls: false
+                    allowEmpty: true
+                - topicNames:
+                  - two
+                  keyRule:
+                    allowNulls: false
+                    allowEmpty: true
+                """);
+
+        TopicMatchingRecordValidationRule ruleOne = new TopicMatchingRecordValidationRule(Set.of("one"), null,
+                new BytebufValidation(new SyntacticallyCorrectJsonConfig(true), false, true));
+        TopicMatchingRecordValidationRule ruleTwo = new TopicMatchingRecordValidationRule(Set.of("two"), new BytebufValidation(null, false, true), null);
+        ValidationConfig expected = new ValidationConfig(true, List.of(ruleOne, ruleTwo), new RecordValidationRule(null, new BytebufValidation(null, false, true)));
+        assertEquals(expected, deserialised);
+    }
+
+}
\ No newline at end of file
diff --git a/kroxylicious-schema-validation/src/test/java/io/kroxylicious/proxy/filter/schema/validation/bytebuf/NullEmptyBytebufValidatorTest.java b/kroxylicious-schema-validation/src/test/java/io/kroxylicious/proxy/filter/schema/validation/bytebuf/NullEmptyBytebufValidatorTest.java
new file mode 100644
index 00000000..337ef005
--- /dev/null
+++ b/kroxylicious-schema-validation/src/test/java/io/kroxylicious/proxy/filter/schema/validation/bytebuf/NullEmptyBytebufValidatorTest.java
@@ -0,0 +1,86 @@
+/*
+ * 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.filter.schema.validation.bytebuf;
+
+import java.nio.ByteBuffer;
+
+import org.apache.kafka.common.record.Record;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import io.kroxylicious.proxy.filter.schema.validation.Result;
+
+import static io.kroxylicious.proxy.filter.schema.validation.bytebuf.BytebufValidators.nullEmptyValidator;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.anyInt;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
+
+class NullEmptyBytebufValidatorTest {
+
+    private final Record record = Mockito.mock(Record.class);
+
+    @Test
+    public void testNullValid() {
+        BytebufValidator mockValidator = mock(BytebufValidator.class);
+        boolean nullValid = true;
+        BytebufValidator validator = nullEmptyValidator(nullValid, true, mockValidator);
+        Result validate = validator.validate(null, 0, record, true);
+        assertTrue(validate.valid());
+        verifyNoInteractions(mockValidator);
+    }
+
+    @Test
+    public void testNullInvalid() {
+        BytebufValidator mockValidator = mock(BytebufValidator.class);
+        boolean nullValid = false;
+        BytebufValidator validator = nullEmptyValidator(nullValid, true, mockValidator);
+        Result validate = validator.validate(null, 0, record, true);
+        assertFalse(validate.valid());
+        verifyNoInteractions(mockValidator);
+    }
+
+    @Test
+    public void testEmptyValid() {
+        BytebufValidator mockValidator = mock(BytebufValidator.class);
+        boolean emptyValid = true;
+        BytebufValidator validator = nullEmptyValidator(true, emptyValid, mockValidator);
+        Result validate = validator.validate(ByteBuffer.wrap(new byte[0]), 0, record, true);
+        assertTrue(validate.valid());
+        verifyNoInteractions(mockValidator);
+    }
+
+    @Test
+    public void testEmptyInvalid() {
+        BytebufValidator mockValidator = mock(BytebufValidator.class);
+        boolean emptyValid = false;
+        BytebufValidator validator = nullEmptyValidator(true, emptyValid, mockValidator);
+        Result validate = validator.validate(ByteBuffer.wrap(new byte[0]), 0, record, true);
+        assertFalse(validate.valid());
+        verifyNoInteractions(mockValidator);
+    }
+
+    @Test
+    public void testDelegation() {
+        BytebufValidator mockValidator = mock(BytebufValidator.class);
+        when(mockValidator.validate(any(), anyInt(), any(), anyBoolean())).thenReturn(new Result(false, "FAIL"));
+        BytebufValidator validator = nullEmptyValidator(true, true, mockValidator);
+        ByteBuffer buffer = ByteBuffer.wrap(new byte[1]);
+        int length = 1;
+        Result validate = validator.validate(buffer, length, record, true);
+        assertFalse(validate.valid());
+        assertEquals("FAIL", validate.errorMessage());
+        verify(mockValidator).validate(buffer, length, record, true);
+    }
+
+}
diff --git a/kroxylicious-schema-validation/src/test/java/io/kroxylicious/proxy/filter/schema/validation/record/KeyAndValueRecordValidatorTest.java b/kroxylicious-schema-validation/src/test/java/io/kroxylicious/proxy/filter/schema/validation/record/KeyAndValueRecordValidatorTest.java
new file mode 100644
index 00000000..54e31fc9
--- /dev/null
+++ b/kroxylicious-schema-validation/src/test/java/io/kroxylicious/proxy/filter/schema/validation/record/KeyAndValueRecordValidatorTest.java
@@ -0,0 +1,58 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.proxy.filter.schema.validation.record;
+
+import org.apache.kafka.common.record.Record;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import io.kroxylicious.proxy.filter.schema.validation.Result;
+import io.kroxylicious.proxy.filter.schema.validation.bytebuf.BytebufValidator;
+
+import static io.kroxylicious.proxy.filter.schema.validation.record.KeyAndValueRecordValidator.keyAndValueValidator;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class KeyAndValueRecordValidatorTest {
+
+    public static final String FAIL_MESSAGE = "fail";
+    public static final BytebufValidator INVALID = (buffer, length, record, isKey) -> new Result(false, FAIL_MESSAGE);
+    public static final BytebufValidator VALID = (buffer, length, record, isKey) -> Result.VALID;
+
+    @Test
+    void testInvalidKey() {
+        RecordValidator recordValidator = keyAndValueValidator(INVALID, VALID);
+        Result validate = recordValidator.validate(Mockito.mock(Record.class));
+        assertFalse(validate.valid());
+        assertEquals("Key was invalid: " + FAIL_MESSAGE, validate.errorMessage());
+    }
+
+    @Test
+    void testInvalidValue() {
+        RecordValidator recordValidator = keyAndValueValidator(VALID, INVALID);
+        Result validate = recordValidator.validate(Mockito.mock(Record.class));
+        assertFalse(validate.valid());
+        assertEquals("Value was invalid: " + FAIL_MESSAGE, validate.errorMessage());
+    }
+
+    @Test
+    void testInvalidKeyAndValue() {
+        RecordValidator recordValidator = keyAndValueValidator(INVALID, INVALID);
+        Result validate = recordValidator.validate(Mockito.mock(Record.class));
+        assertFalse(validate.valid());
+        assertEquals("Key was invalid: " + FAIL_MESSAGE, validate.errorMessage());
+    }
+
+    @Test
+    void testValidKeyAndValue() {
+        RecordValidator recordValidator = keyAndValueValidator(VALID, VALID);
+        Result validate = recordValidator.validate(Mockito.mock(Record.class));
+        assertTrue(validate.valid());
+    }
+
+}
\ No newline at end of file