Repository: kroxylicious/kroxylicious (java)

Issue:
Record Validation Filter: zero-ack produce requests combined with partial batch forwarding causes memory leak and OOM
**Describe the bug**

If I have a RecordValidationFilter with `forwardPartialRequests` enabled, and a producer that is producing lots of zero-ack produce requests we are leaking memory.

1. We put data in a correlation map about the failed topic partitions and then expect to augment them back into the response.
2. In the zero-ack produce request case there is no response to augment and the correlated data sits in the map forever
3. This can grow until OOM

**To Reproduce**

1. run up a kafka broker on 9092

Configure a filter that blocks null values on all topics with forwardPartialRequests enabled

```yaml
adminHttp:
  endpoints:
    prometheus: {}
virtualClusters:
  demo:
    targetCluster:
      bootstrap_servers: localhost:9092
    clusterNetworkAddressConfigProvider:
      type: PortPerBrokerClusterNetworkAddressConfigProvider
      config:
        bootstrapAddress: localhost:9192
    logNetwork: false
    logFrames: false
filters:
- type: RecordValidation
  config:
      rules: []
      defaultRule:
        valueRule:
          allowNulls: false                                                 
          allowEmpty: true
      forwardPartialRequests: true
```

hammer the proxy with zero-ack produce requests where some partitions contain only invalid records and some partitions contain only valid records to prompt the partial request forwarding logic to kick in.

```
package org.example;

import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.List;
import java.util.Map;

public class Main {
    public static void main(String[] args) {
        Admin admin = Admin.create(Map.of(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"));
        int numPartitions = 20;
        String topic = "test5";
        try {
            admin.createTopics(List.of(new NewTopic(topic, numPartitions, (short) 1))).all().get();
        } catch (Exception e) {
            System.out.println(e);
        }
        KafkaProducer<String, String> producer = new KafkaProducer<>(Map.of(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9192", ProducerConfig.LINGER_MS_CONFIG, 1000, ProducerConfig.ACKS_CONFIG, "0"), new StringSerializer(), new StringSerializer());
        int i = 0;
        while(true) {
            int partition = i++ % numPartitions;
            producer.send(new ProducerRecord<>(topic, partition, "key", partition % 2 == 0 ? null : "it's present"));
        }
    }
}
```

observe Kroxylicious memory in jconsole or equivalent head to the heap limit and requests eventually start to slow down to a crawl:
![Screenshot from 2024-08-16 15-45-09](https://github.com/user-attachments/assets/d7f977e4-1023-4b97-aee3-9f690d054248)

**Expected behavior**

Not a big old memory leak, that's for sure.

It's unclear what record validation bejaviour should be in the context of zero-ack requests. The producer is firing and forgetting, with no response from the broker expected. We still want to prevent the bad data hitting the broker. Maybe we need some metrics to let the user know we are preventing writes.

We also need to ensure we do not respond to zero-ack requests. Currently we respond from the filter if we invalidate the whole ProduceRequest because all topic-partitions contain invalid records or we are transactional and any record in the request was invalid.

Pull request:
Fix #1436 : remove forwardPartialRequests feature of record validation filter

### Type of change

_Select the type of your PR_

- Bugfix

### Description

Why: in the commentary of #1436 it was concluded that the inclusion of `forwardPartialRequests` option in the RecordValidation filter was an error. The problem being that if a client is sending incorrect data, then stopping immediately is the safe course of action.   Accepting some data and rejecting other is likely to lead to more harm than good.

We said we wanted to drop the feature (without deprecation).


### Additional Context

_Why are you making this pull request?_

### Checklist

_Please go through this checklist and make sure all applicable tasks have been done_

- [x] Write tests
- [x] Update documentation
- [ ] Make sure all unit/integration tests pass
- [x] Make sure all Sonarcloud warnings are addressed or are justifiably ignored.
- [ ] If applicable to the change, [trigger the system test suite](../blob/main/DEV_GUIDE.md#jenkins-pipeline-for-system-tests).  Make sure tests pass.
- [ ] If applicable to the change, [trigger the performance test suite](../blob/main/PERFORMANCE.md#jenkins-pipeline-for-performance). Ensure that any degradations to performance numbers are understood and justified.
- [ ] Ensure the PR references relevant issue(s) so they are closed on merging.
- [ ] For user facing changes, update CHANGELOG.md (remember to include changes affecting the API of the test artefacts too).

> **_NOTE:_**  You must be a member of `@kroxylicious/developers` to trigger the system test and performance test suites.  If you are not part of this group, comment on the PR requesting a trigger, tagging `@kroxylicious/developers`.


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index e5dfad91..194fcd2c 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,6 +7,7 @@ Format `<github issue/pr number>: <short description>`.
 
 ## SNAPSHOT
 
+* [#1657](https://github.com/kroxylicious/kroxylicious/pull/1657) Remove forwardPartialRequests feature of record validation filter
 * [#1635](https://github.com/kroxylicious/kroxylicious/pull/1635) Handle ApiVersions unsupported version downgrade
 * [#1648](https://github.com/kroxylicious/kroxylicious/pull/1648) Add test-only feature mechanism to Proxy configuration
 * [#1379](https://github.com/kroxylicious/kroxylicious/issues/1379) Remove Deprecated EnvelopeEncryption
@@ -22,7 +23,8 @@ Format `<github issue/pr number>: <short description>`.
 
 ### Changes, deprecations and removals
 
-* The deprecated EnvelopeEncryption filter is now removed.  Use RecordEncryption instead. 
+* The deprecated EnvelopeEncryption filter is now removed.  Use RecordEncryption instead.
+* The deprecated forwardPartialRequests option has been removed from the Record Validation Filter.
 
 ## 0.8.0
 
diff --git a/docs/modules/record-validation/proc-record-validation.adoc b/docs/modules/record-validation/proc-record-validation.adoc
index 5cda4a69..2cfd6e5d 100644
--- a/docs/modules/record-validation/proc-record-validation.adoc
+++ b/docs/modules/record-validation/proc-record-validation.adoc
@@ -37,18 +37,11 @@ filters:
             <rule definition>                                          # <2>
           valueRule:
             <rule definition>                                          # <3>
-        forwardPartialRequests: false                                  # <5>
 ----
 <1> List of topic names to which the validation rules will be applied.
 <2> Validation rules that are applied to the record's key.
 <3> Validation rules that are applied to the record's value.
 <4> (Optional) Default rule that is applied to any topics for which there is no explict rule defined.
-<5> If `false`, any record failing validation will cause the entire produce request to be rejected.
-    If `true`, only partitions within the request where all records validated successfully will be forwarded to the
-    broker. The response the client will receive will be a mixed response with response returned from the broker
-    for the partitions whose records passed validation and error responses for the partitions that failed validation.
-    If the client is using transactions, this setting is *ignored*. Any failing record validation will
-    cause the entire produce request to be rejected.
 
 Replace the token `<rule definition>`  in the YAML configuration with either a Schema Validation rule or a JSON Syntax Validation rule depending on your requirements.
 
diff --git a/kroxylicious-filters/kroxylicious-record-validation/src/main/java/io/kroxylicious/proxy/filter/validation/ProduceRequestValidatorBuilder.java b/kroxylicious-filters/kroxylicious-record-validation/src/main/java/io/kroxylicious/proxy/filter/validation/ProduceRequestValidatorBuilder.java
index c1175cfc..73110e84 100644
--- a/kroxylicious-filters/kroxylicious-record-validation/src/main/java/io/kroxylicious/proxy/filter/validation/ProduceRequestValidatorBuilder.java
+++ b/kroxylicious-filters/kroxylicious-record-validation/src/main/java/io/kroxylicious/proxy/filter/validation/ProduceRequestValidatorBuilder.java
@@ -38,8 +38,8 @@ class ProduceRequestValidatorBuilder {
      */
     static ProduceRequestValidator build(ValidationConfig config) {
         RoutingProduceRequestValidator.RoutingProduceRequestValidatorBuilder builder = RoutingProduceRequestValidator.builder();
-        config.getRules().forEach(rule -> builder.appendValidatorForTopicPattern(rule.getTopicNames(), toValidatorWithNullHandling(rule)));
-        RecordValidationRule defaultRule = config.getDefaultRule();
+        config.rules().forEach(rule -> builder.appendValidatorForTopicPattern(rule.getTopicNames(), toValidatorWithNullHandling(rule)));
+        RecordValidationRule defaultRule = config.defaultRule();
         TopicValidator defaultValidator = defaultRule == null ? TopicValidators.allValid() : toValidatorWithNullHandling(defaultRule);
         builder.setDefaultValidator(defaultValidator);
         return builder.build();
diff --git a/kroxylicious-filters/kroxylicious-record-validation/src/main/java/io/kroxylicious/proxy/filter/validation/RecordValidation.java b/kroxylicious-filters/kroxylicious-record-validation/src/main/java/io/kroxylicious/proxy/filter/validation/RecordValidation.java
index 72e6e5a8..e0d7796b 100644
--- a/kroxylicious-filters/kroxylicious-record-validation/src/main/java/io/kroxylicious/proxy/filter/validation/RecordValidation.java
+++ b/kroxylicious-filters/kroxylicious-record-validation/src/main/java/io/kroxylicious/proxy/filter/validation/RecordValidation.java
@@ -24,7 +24,7 @@ public class RecordValidation implements FilterFactory<ValidationConfig, Validat
     @Override
     public RecordValidationFilter createFilter(FilterFactoryContext context, ValidationConfig configuration) {
         ProduceRequestValidator validator = ProduceRequestValidatorBuilder.build(configuration);
-        return new RecordValidationFilter(configuration.isForwardPartialRequests(), validator);
+        return new RecordValidationFilter(validator);
     }
 
 }
diff --git a/kroxylicious-filters/kroxylicious-record-validation/src/main/java/io/kroxylicious/proxy/filter/validation/RecordValidationFilter.java b/kroxylicious-filters/kroxylicious-record-validation/src/main/java/io/kroxylicious/proxy/filter/validation/RecordValidationFilter.java
index ac80efb2..b86993e5 100644
--- a/kroxylicious-filters/kroxylicious-record-validation/src/main/java/io/kroxylicious/proxy/filter/validation/RecordValidationFilter.java
+++ b/kroxylicious-filters/kroxylicious-record-validation/src/main/java/io/kroxylicious/proxy/filter/validation/RecordValidationFilter.java
@@ -31,37 +31,25 @@ import io.kroxylicious.proxy.filter.validation.validators.topic.RecordValidation
 import io.kroxylicious.proxy.filter.validation.validators.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>
+ * The filter intercepts the produce requests and subject the records contained within them to validation. If the
+ * validation fails, the whole produce request is rejected and the producing application receives an error
+ * response {@link Errors#INVALID_RECORD}.  The broker does not receive rejected produce requests.
  */
 public class RecordValidationFilter implements ProduceRequestFilter, ProduceResponseFilter {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(RecordValidationFilter.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 RecordValidationFilter(boolean forwardPartialRequests, ProduceRequestValidator validator) {
+    public RecordValidationFilter(ProduceRequestValidator validator) {
         if (validator == null) {
             throw new IllegalArgumentException("validator is null");
         }
-        this.forwardPartialRequests = forwardPartialRequests;
         this.validator = validator;
     }
 
@@ -70,7 +58,7 @@ public class RecordValidationFilter implements ProduceRequestFilter, ProduceResp
         CompletionStage<ProduceRequestValidationResult> validationStage = validator.validateRequest(request);
         return validationStage.thenCompose(result -> {
             if (result.isAnyTopicPartitionInvalid()) {
-                return handleInvalidTopicPartitions(header, request, context, result);
+                return handleInvalidTopicPartitions(request, context, result);
             }
             else {
                 return context.forwardRequest(header, request);
@@ -78,29 +66,11 @@ public class RecordValidationFilter implements ProduceRequestFilter, ProduceResp
         });
     }
 
-    private CompletionStage<RequestFilterResult> handleInvalidTopicPartitions(RequestHeaderData header, ProduceRequestData request, FilterContext context,
+    private CompletionStage<RequestFilterResult> handleInvalidTopicPartitions(ProduceRequestData request, FilterContext context,
                                                                               ProduceRequestValidationResult result) {
-        if (result.isAllTopicPartitionsInvalid()) {
-            LOGGER.debug("all topic-partitions for request contained invalid data: {}", result);
-            ProduceResponseData response = invalidateEntireRequest(request, result);
-            return context.requestFilterResultBuilder().shortCircuitResponse(response).completed();
-        }
-        // 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);
-            return 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);
-            return context.requestFilterResultBuilder().shortCircuitResponse(response).completed();
-        }
+        LOGGER.debug("At least one topic-partitions with the request contained invalid records: {}. Produce request will be rejected.", result);
+        ProduceResponseData response = invalidateEntireRequest(request, result);
+        return context.requestFilterResultBuilder().shortCircuitResponse(response).completed();
     }
 
     private static ProduceResponseData invalidateEntireRequest(ProduceRequestData request, ProduceRequestValidationResult produceRequestValidationResult) {
diff --git a/kroxylicious-filters/kroxylicious-record-validation/src/main/java/io/kroxylicious/proxy/filter/validation/config/ValidationConfig.java b/kroxylicious-filters/kroxylicious-record-validation/src/main/java/io/kroxylicious/proxy/filter/validation/config/ValidationConfig.java
index e98f2427..83124739 100644
--- a/kroxylicious-filters/kroxylicious-record-validation/src/main/java/io/kroxylicious/proxy/filter/validation/config/ValidationConfig.java
+++ b/kroxylicious-filters/kroxylicious-record-validation/src/main/java/io/kroxylicious/proxy/filter/validation/config/ValidationConfig.java
@@ -7,90 +7,23 @@
 package io.kroxylicious.proxy.filter.validation.config;
 
 import java.util.List;
-import java.util.Objects;
 
-import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
 
 /**
  * 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)
+ * the data for all topic-partitions within a ProduceRequest.
+ *
+ * @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
  */
-public class ValidationConfig {
-
-    /**
-     * 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);
-    }
+public record ValidationConfig(@JsonProperty("rules") List<TopicMatchingRecordValidationRule> rules,
+                               @JsonProperty("defaultRule") RecordValidationRule defaultRule) {
 
     @Override
     public String toString() {
         return "ValidationConfig{" +
-                "forwardPartialRequests=" + forwardPartialRequests +
-                ", rules=" + rules +
+                "rules=" + rules +
                 ", defaultRule=" + defaultRule +
                 '}';
     }
diff --git a/kroxylicious-filters/kroxylicious-record-validation/src/main/java/io/kroxylicious/proxy/filter/validation/validators/request/ProduceRequestValidationResult.java b/kroxylicious-filters/kroxylicious-record-validation/src/main/java/io/kroxylicious/proxy/filter/validation/validators/request/ProduceRequestValidationResult.java
index 32c3a714..7b42a190 100644
--- a/kroxylicious-filters/kroxylicious-record-validation/src/main/java/io/kroxylicious/proxy/filter/validation/validators/request/ProduceRequestValidationResult.java
+++ b/kroxylicious-filters/kroxylicious-record-validation/src/main/java/io/kroxylicious/proxy/filter/validation/validators/request/ProduceRequestValidationResult.java
@@ -9,7 +9,6 @@ package io.kroxylicious.proxy.filter.validation.validators.request;
 import java.util.Map;
 import java.util.stream.Stream;
 
-import io.kroxylicious.proxy.filter.validation.validators.topic.PartitionValidationResult;
 import io.kroxylicious.proxy.filter.validation.validators.topic.TopicValidationResult;
 
 /**
@@ -27,49 +26,6 @@ public record ProduceRequestValidationResult(Map<String, TopicValidationResult>
         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
diff --git a/kroxylicious-filters/kroxylicious-record-validation/src/main/java/io/kroxylicious/proxy/filter/validation/validators/topic/AllValidTopicValidationResult.java b/kroxylicious-filters/kroxylicious-record-validation/src/main/java/io/kroxylicious/proxy/filter/validation/validators/topic/AllValidTopicValidationResult.java
index a5fd5be0..170698cb 100644
--- a/kroxylicious-filters/kroxylicious-record-validation/src/main/java/io/kroxylicious/proxy/filter/validation/validators/topic/AllValidTopicValidationResult.java
+++ b/kroxylicious-filters/kroxylicious-record-validation/src/main/java/io/kroxylicious/proxy/filter/validation/validators/topic/AllValidTopicValidationResult.java
@@ -16,11 +16,6 @@ record AllValidTopicValidationResult(String topicName) implements TopicValidatio
         return false;
     }
 
-    @Override
-    public boolean isAllPartitionsInvalid() {
-        return false;
-    }
-
     @Override
     public Stream<PartitionValidationResult> invalidPartitions() {
         return Stream.empty();
diff --git a/kroxylicious-filters/kroxylicious-record-validation/src/main/java/io/kroxylicious/proxy/filter/validation/validators/topic/PerPartitionTopicValidationResult.java b/kroxylicious-filters/kroxylicious-record-validation/src/main/java/io/kroxylicious/proxy/filter/validation/validators/topic/PerPartitionTopicValidationResult.java
index 7e5f505d..c8344d87 100644
--- a/kroxylicious-filters/kroxylicious-record-validation/src/main/java/io/kroxylicious/proxy/filter/validation/validators/topic/PerPartitionTopicValidationResult.java
+++ b/kroxylicious-filters/kroxylicious-record-validation/src/main/java/io/kroxylicious/proxy/filter/validation/validators/topic/PerPartitionTopicValidationResult.java
@@ -16,14 +16,6 @@ record PerPartitionTopicValidationResult(String topicName, Map<Integer, Partitio
         return partitionValidationResults.values().stream().anyMatch(partitionValidationResult -> !partitionValidationResult.allRecordsValid());
     }
 
-    @Override
-    public boolean isAllPartitionsInvalid() {
-        if (partitionValidationResults.isEmpty()) {
-            return false;
-        }
-        return partitionValidationResults.values().stream().noneMatch(PartitionValidationResult::allRecordsValid);
-    }
-
     @Override
     public Stream<PartitionValidationResult> invalidPartitions() {
         return partitionValidationResults.values().stream().filter(partitionValidationResult -> !partitionValidationResult.allRecordsValid());
diff --git a/kroxylicious-filters/kroxylicious-record-validation/src/main/java/io/kroxylicious/proxy/filter/validation/validators/topic/TopicValidationResult.java b/kroxylicious-filters/kroxylicious-record-validation/src/main/java/io/kroxylicious/proxy/filter/validation/validators/topic/TopicValidationResult.java
index 2a4dbaa3..a3cdbe2b 100644
--- a/kroxylicious-filters/kroxylicious-record-validation/src/main/java/io/kroxylicious/proxy/filter/validation/validators/topic/TopicValidationResult.java
+++ b/kroxylicious-filters/kroxylicious-record-validation/src/main/java/io/kroxylicious/proxy/filter/validation/validators/topic/TopicValidationResult.java
@@ -19,12 +19,6 @@ public interface TopicValidationResult {
      */
     boolean isAnyPartitionInvalid();
 
-    /**
-     * are all partitions invalid
-     * @return true if all partitions invalid
-     */
-    boolean isAllPartitionsInvalid();
-
     /**
      * get invalid partitions
      * @return stream of invalid partitions


Test patch:
diff --git a/kroxylicious-filters/kroxylicious-record-validation/src/test/java/io/kroxylicious/proxy/filter/validation/RecordValidationFilterTest.java b/kroxylicious-filters/kroxylicious-record-validation/src/test/java/io/kroxylicious/proxy/filter/validation/RecordValidationFilterTest.java
index 955fc189..aefa9380 100644
--- a/kroxylicious-filters/kroxylicious-record-validation/src/test/java/io/kroxylicious/proxy/filter/validation/RecordValidationFilterTest.java
+++ b/kroxylicious-filters/kroxylicious-record-validation/src/test/java/io/kroxylicious/proxy/filter/validation/RecordValidationFilterTest.java
@@ -91,14 +91,14 @@ class RecordValidationFilterTest {
     @SuppressWarnings("DataFlowIssue")
     @Test
     void rejectsNullValidator() {
-        assertThatThrownBy(() -> new RecordValidationFilter(false, null))
+        assertThatThrownBy(() -> new RecordValidationFilter(null))
                 .isInstanceOf(IllegalArgumentException.class);
     }
 
     @Test
     void requestThatPassesValidationIsForwarded() {
         // Given
-        var validator = new RecordValidationFilter(false, produceRequestValidator);
+        var validator = new RecordValidationFilter(produceRequestValidator);
 
         when(topicValidationResult.isAnyPartitionInvalid()).thenReturn(false);
 
@@ -125,10 +125,9 @@ class RecordValidationFilterTest {
     @Test
     void requestWithAllPartitionsFailedIsRejectedWithShortCircuitResponse() {
         // Given
-        var validator = new RecordValidationFilter(false, produceRequestValidator);
+        var validator = new RecordValidationFilter(produceRequestValidator);
 
         when(topicValidationResult.isAnyPartitionInvalid()).thenReturn(true);
-        when(topicValidationResult.isAllPartitionsInvalid()).thenReturn(true);
         when(topicValidationResult.getPartitionResult(0)).thenReturn(new PartitionValidationResult(0, List.of(new RecordValidationFailure(0, "record error"))));
 
         var header = new RequestHeaderData();
@@ -161,60 +160,12 @@ class RecordValidationFilterTest {
                 });
     }
 
-    @Test
-    void requestWithSomePartitionsFailedIsIsForwarded() {
-        // Given
-        var validator = new RecordValidationFilter(true, produceRequestValidator);
-
-        when(topicValidationResult.isAnyPartitionInvalid()).thenReturn(true);
-        when(topicValidationResult.isAllPartitionsInvalid()).thenReturn(false);
-        when(topicValidationResult.getPartitionResult(0)).thenReturn(new PartitionValidationResult(0, List.of(new RecordValidationFailure(0, "record error"))));
-        when(topicValidationResult.getPartitionResult(1)).thenReturn(new PartitionValidationResult(1, List.of()));
-
-        var header = new RequestHeaderData();
-        final ProduceRequestData.PartitionProduceData partition1 = new ProduceRequestData.PartitionProduceData();
-        final ProduceRequestData.PartitionProduceData partition2 = new ProduceRequestData.PartitionProduceData();
-        partition2.setIndex(1);
-        var request = buildProduceRequestData(new ProduceRequestData.TopicProduceData()
-                .setName(MY_TOPIC)
-                .setPartitionData(
-                        new ArrayList<>(List.of(
-                                partition1,
-                                partition2))));
-        when(produceRequestValidator.validateRequest(request)).thenReturn(
-                CompletableFuture.completedStage(new ProduceRequestValidationResult(Map.of(MY_TOPIC, topicValidationResult))));
-
-        // When
-        var result = validator.onProduceRequest(HIGHEST_SUPPORTED_VERSION, header, request, context);
-
-        // Then
-        assertThat(result)
-                .succeedsWithin(Duration.ofSeconds(1))
-                .satisfies(rfr -> {
-                    assertThat(rfr.shortCircuitResponse()).isFalse();
-                    assertThat(rfr.message())
-                            .isInstanceOf(ProduceRequestData.class);
-
-                    var prd = (ProduceRequestData) rfr.message();
-                    assertThat(prd.topicData())
-                            .singleElement()
-                            .satisfies(tpr -> {
-                                assertThat(tpr.name()).isEqualTo(MY_TOPIC);
-                                assertThat(tpr.partitionData())
-                                        .singleElement()
-                                        .extracting(ProduceRequestData.PartitionProduceData::index)
-                                        .isEqualTo(1);
-                            });
-                });
-    }
-
     @Test
     void requestWithSomePartitionsFailedIsRejected() {
         // Given
-        var validator = new RecordValidationFilter(false, produceRequestValidator);
+        var validator = new RecordValidationFilter(produceRequestValidator);
 
         when(topicValidationResult.isAnyPartitionInvalid()).thenReturn(true);
-        when(topicValidationResult.isAllPartitionsInvalid()).thenReturn(false);
         when(topicValidationResult.getPartitionResult(0)).thenReturn(new PartitionValidationResult(0, List.of(new RecordValidationFailure(0, "record error"))));
         when(topicValidationResult.getPartitionResult(1)).thenReturn(new PartitionValidationResult(1, List.of()));
 
@@ -261,10 +212,9 @@ class RecordValidationFilterTest {
     @Test
     void requestWithAllPartitionsFailedIsRejectedWithShortCircuitResponseInTransaction() {
         // Given
-        var validator = new RecordValidationFilter(true, produceRequestValidator);
+        var validator = new RecordValidationFilter(produceRequestValidator);
 
         when(topicValidationResult.isAnyPartitionInvalid()).thenReturn(true);
-        when(topicValidationResult.isAllPartitionsInvalid()).thenReturn(true);
         when(topicValidationResult.getPartitionResult(0)).thenReturn(new PartitionValidationResult(0, List.of(new RecordValidationFailure(0, "record error"))));
 
         var header = new RequestHeaderData();
@@ -300,10 +250,9 @@ class RecordValidationFilterTest {
     @Test
     void requestWithSomePartitionsFailedIsRejectedWithShortCircuitResponseInTransaction() {
         // Given
-        var validator = new RecordValidationFilter(true, produceRequestValidator);
+        var validator = new RecordValidationFilter(produceRequestValidator);
 
         when(topicValidationResult.isAnyPartitionInvalid()).thenReturn(true);
-        when(topicValidationResult.isAllPartitionsInvalid()).thenReturn(false);
         when(topicValidationResult.getPartitionResult(0)).thenReturn(new PartitionValidationResult(0, List.of(new RecordValidationFailure(0, "record error"))));
 
         var header = new RequestHeaderData();
diff --git a/kroxylicious-filters/kroxylicious-record-validation/src/test/java/io/kroxylicious/proxy/filter/validation/RecordValidationTest.java b/kroxylicious-filters/kroxylicious-record-validation/src/test/java/io/kroxylicious/proxy/filter/validation/RecordValidationTest.java
index 0f3a3606..fe9c4df4 100644
--- a/kroxylicious-filters/kroxylicious-record-validation/src/test/java/io/kroxylicious/proxy/filter/validation/RecordValidationTest.java
+++ b/kroxylicious-filters/kroxylicious-record-validation/src/test/java/io/kroxylicious/proxy/filter/validation/RecordValidationTest.java
@@ -30,7 +30,7 @@ class RecordValidationTest {
     @Test
     void shouldInitAndCreateFilter() {
         RecordValidation factory = new RecordValidation();
-        var config = factory.initialize(null, new ValidationConfig(true, List.of(), new RecordValidationRule(null, null)));
+        var config = factory.initialize(null, new ValidationConfig(List.of(), new RecordValidationRule(null, null)));
         Filter filter = factory.createFilter(null, config);
         assertThat(filter).isNotNull().isInstanceOf(RecordValidationFilter.class);
     }
diff --git a/kroxylicious-filters/kroxylicious-record-validation/src/test/java/io/kroxylicious/proxy/filter/validation/config/ValidationConfigTest.java b/kroxylicious-filters/kroxylicious-record-validation/src/test/java/io/kroxylicious/proxy/filter/validation/config/ValidationConfigTest.java
index 899617d8..010778fe 100644
--- a/kroxylicious-filters/kroxylicious-record-validation/src/test/java/io/kroxylicious/proxy/filter/validation/config/ValidationConfigTest.java
+++ b/kroxylicious-filters/kroxylicious-record-validation/src/test/java/io/kroxylicious/proxy/filter/validation/config/ValidationConfigTest.java
@@ -40,7 +40,7 @@ class ValidationConfigTest {
         TopicMatchingRecordValidationRule ruleOne = new TopicMatchingRecordValidationRule(Set.of("one"), null,
                 new BytebufValidation(new SyntacticallyCorrectJsonConfig(false), null, true, false));
         TopicMatchingRecordValidationRule ruleTwo = new TopicMatchingRecordValidationRule(Set.of("two"), new BytebufValidation(null, null, true, false), null);
-        ValidationConfig expected = new ValidationConfig(false, List.of(ruleOne, ruleTwo),
+        ValidationConfig expected = new ValidationConfig(List.of(ruleOne, ruleTwo),
                 new RecordValidationRule(null, new BytebufValidation(null, null, true, false)));
         assertEquals(expected, deserialised);
     }
@@ -49,7 +49,6 @@ class ValidationConfigTest {
     void testDecodeNonDefaultValues() throws JsonProcessingException {
         ObjectMapper yamlMapper = new ObjectMapper(new YAMLFactory());
         ValidationConfig deserialised = yamlMapper.readerFor(ValidationConfig.class).readValue("""
-                forwardPartialRequests: true
                 defaultRule:
                   valueRule:
                     allowNulls: false
@@ -72,7 +71,7 @@ class ValidationConfigTest {
         TopicMatchingRecordValidationRule ruleOne = new TopicMatchingRecordValidationRule(Set.of("one"), null,
                 new BytebufValidation(new SyntacticallyCorrectJsonConfig(true), null, false, true));
         TopicMatchingRecordValidationRule ruleTwo = new TopicMatchingRecordValidationRule(Set.of("two"), new BytebufValidation(null, null, false, true), null);
-        ValidationConfig expected = new ValidationConfig(true, List.of(ruleOne, ruleTwo), new RecordValidationRule(null, new BytebufValidation(null, null, false, true)));
+        ValidationConfig expected = new ValidationConfig(List.of(ruleOne, ruleTwo), new RecordValidationRule(null, new BytebufValidation(null, null, false, true)));
         assertEquals(expected, deserialised);
     }
 
@@ -95,7 +94,7 @@ class ValidationConfigTest {
         TopicMatchingRecordValidationRule ruleOne = new TopicMatchingRecordValidationRule(Set.of("one"), null,
                 new BytebufValidation(new SyntacticallyCorrectJsonConfig(false), null, true, false));
         TopicMatchingRecordValidationRule ruleTwo = new TopicMatchingRecordValidationRule(Set.of("two"), new BytebufValidation(null, null, true, false), null);
-        ValidationConfig expected = new ValidationConfig(false, List.of(ruleOne, ruleTwo),
+        ValidationConfig expected = new ValidationConfig(List.of(ruleOne, ruleTwo),
                 new RecordValidationRule(null, new BytebufValidation(null, null, true, false)));
         assertEquals(expected, deserialised);
     }
@@ -104,7 +103,6 @@ class ValidationConfigTest {
     void testDecodeNonDefaultValuesSchemaValidation() throws JsonProcessingException, MalformedURLException {
         ObjectMapper yamlMapper = new ObjectMapper(new YAMLFactory());
         ValidationConfig deserialised = yamlMapper.readerFor(ValidationConfig.class).readValue("""
-                forwardPartialRequests: true
                 defaultRule:
                   valueRule:
                     allowNulls: false
@@ -131,7 +129,7 @@ class ValidationConfigTest {
                 new BytebufValidation(new SyntacticallyCorrectJsonConfig(true), new SchemaValidationConfig(URI.create("http://localhost:8080").toURL(), 1L), false,
                         true));
         TopicMatchingRecordValidationRule ruleTwo = new TopicMatchingRecordValidationRule(Set.of("two"), new BytebufValidation(null, null, false, true), null);
-        ValidationConfig expected = new ValidationConfig(true, List.of(ruleOne, ruleTwo), new RecordValidationRule(null, new BytebufValidation(null, null, false, true)));
+        ValidationConfig expected = new ValidationConfig(List.of(ruleOne, ruleTwo), new RecordValidationRule(null, new BytebufValidation(null, null, false, true)));
         assertEquals(expected, deserialised);
     }
 
@@ -139,7 +137,6 @@ class ValidationConfigTest {
     void testDecodeInvalidValuesSchemaValidation() throws JsonProcessingException {
         ObjectMapper yamlMapper = new ObjectMapper(new YAMLFactory());
         ValidationConfig deserialised = yamlMapper.readerFor(ValidationConfig.class).readValue("""
-                forwardPartialRequests: true
                 defaultRule:
                   valueRule:
                     allowNulls: false
@@ -162,7 +159,7 @@ class ValidationConfigTest {
         TopicMatchingRecordValidationRule ruleOne = new TopicMatchingRecordValidationRule(Set.of("one"), null,
                 new BytebufValidation(new SyntacticallyCorrectJsonConfig(true), null, false, true));
         TopicMatchingRecordValidationRule ruleTwo = new TopicMatchingRecordValidationRule(Set.of("two"), new BytebufValidation(null, null, false, true), null);
-        ValidationConfig expected = new ValidationConfig(true, List.of(ruleOne, ruleTwo), new RecordValidationRule(null, new BytebufValidation(null, null, false, true)));
+        ValidationConfig expected = new ValidationConfig(List.of(ruleOne, ruleTwo), new RecordValidationRule(null, new BytebufValidation(null, null, false, true)));
         assertEquals(expected, deserialised);
     }
 }
diff --git a/kroxylicious-filters/kroxylicious-record-validation/src/test/java/io/kroxylicious/proxy/filter/validation/validators/topic/PerPartitionTopicValidationResultTest.java b/kroxylicious-filters/kroxylicious-record-validation/src/test/java/io/kroxylicious/proxy/filter/validation/validators/topic/PerPartitionTopicValidationResultTest.java
index 46af380a..b85ac9a1 100644
--- a/kroxylicious-filters/kroxylicious-record-validation/src/test/java/io/kroxylicious/proxy/filter/validation/validators/topic/PerPartitionTopicValidationResultTest.java
+++ b/kroxylicious-filters/kroxylicious-record-validation/src/test/java/io/kroxylicious/proxy/filter/validation/validators/topic/PerPartitionTopicValidationResultTest.java
@@ -23,7 +23,6 @@ class PerPartitionTopicValidationResultTest {
         var result = new PerPartitionTopicValidationResult("foo", Map.of());
 
         assertThat(result)
-                .returns(false, PerPartitionTopicValidationResult::isAllPartitionsInvalid)
                 .returns(false, PerPartitionTopicValidationResult::isAnyPartitionInvalid);
         assertThat(result)
                 .extracting(PerPartitionTopicValidationResult::invalidPartitions, STREAM)
@@ -36,8 +35,7 @@ class PerPartitionTopicValidationResultTest {
                 Map.of(1, new PartitionValidationResult(2, List.of(INVALID))));
 
         assertThat(result)
-                .returns(true, PerPartitionTopicValidationResult::isAnyPartitionInvalid)
-                .returns(true, PerPartitionTopicValidationResult::isAllPartitionsInvalid);
+                .returns(true, PerPartitionTopicValidationResult::isAnyPartitionInvalid);
     }
 
     @Test
@@ -47,8 +45,7 @@ class PerPartitionTopicValidationResultTest {
                         1, new PartitionValidationResult(1, List.of(INVALID))));
 
         assertThat(result)
-                .returns(true, PerPartitionTopicValidationResult::isAnyPartitionInvalid)
-                .returns(false, PerPartitionTopicValidationResult::isAllPartitionsInvalid);
+                .returns(true, PerPartitionTopicValidationResult::isAnyPartitionInvalid);
     }
 
     @Test
diff --git a/kroxylicious-filters/kroxylicious-record-validation/src/test/java/io/kroxylicious/proxy/filter/validation/validators/topic/PerRecordTopicValidatorTest.java b/kroxylicious-filters/kroxylicious-record-validation/src/test/java/io/kroxylicious/proxy/filter/validation/validators/topic/PerRecordTopicValidatorTest.java
index 84aab5b7..e3a8974e 100644
--- a/kroxylicious-filters/kroxylicious-record-validation/src/test/java/io/kroxylicious/proxy/filter/validation/validators/topic/PerRecordTopicValidatorTest.java
+++ b/kroxylicious-filters/kroxylicious-record-validation/src/test/java/io/kroxylicious/proxy/filter/validation/validators/topic/PerRecordTopicValidatorTest.java
@@ -46,7 +46,6 @@ class PerRecordTopicValidatorTest {
         var result = topicValidator.validateTopicData(tpd);
         assertThat(result)
                 .succeedsWithin(Duration.ofSeconds(1))
-                .returns(false, TopicValidationResult::isAllPartitionsInvalid)
                 .returns(false, TopicValidationResult::isAnyPartitionInvalid)
                 .extracting(TopicValidationResult::invalidPartitions, stream(PartitionValidationResult.class))
                 .isEmpty();
@@ -63,7 +62,6 @@ class PerRecordTopicValidatorTest {
         var result = topicValidator.validateTopicData(tpd);
         assertThat(result)
                 .succeedsWithin(Duration.ofSeconds(1))
-                .returns(true, TopicValidationResult::isAllPartitionsInvalid)
                 .returns(true, TopicValidationResult::isAnyPartitionInvalid);
 
         var invalidPartitions = result.toCompletableFuture().join().invalidPartitions();
@@ -118,7 +116,6 @@ class PerRecordTopicValidatorTest {
 
         assertThat(result)
                 .succeedsWithin(Duration.ofSeconds(1))
-                .returns(false, TopicValidationResult::isAllPartitionsInvalid)
                 .returns(true, TopicValidationResult::isAnyPartitionInvalid);
 
         var invalidPartitions = result.toCompletableFuture().join().invalidPartitions();
@@ -149,7 +146,6 @@ class PerRecordTopicValidatorTest {
 
         assertThat(result)
                 .succeedsWithin(Duration.ofSeconds(1))
-                .returns(true, TopicValidationResult::isAllPartitionsInvalid)
                 .returns(true, TopicValidationResult::isAnyPartitionInvalid);
 
         var invalidPartitions = result.toCompletableFuture().join().invalidPartitions().toList();
diff --git a/kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/filter/validation/JsonSyntaxRecordValidationIT.java b/kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/filter/validation/JsonSyntaxRecordValidationIT.java
index 79773f54..c13f246a 100644
--- a/kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/filter/validation/JsonSyntaxRecordValidationIT.java
+++ b/kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/filter/validation/JsonSyntaxRecordValidationIT.java
@@ -12,7 +12,6 @@ import java.util.Map;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.common.InvalidRecordException;
-import org.apache.kafka.common.KafkaException;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 
@@ -39,7 +38,7 @@ class JsonSyntaxRecordValidationIT extends RecordValidationBaseIT {
     @Test
     void invalidJsonProduceRejected(KafkaCluster cluster, Topic topic) {
         var config = proxy(cluster)
-                .addToFilters(createFilterDef(false, topic));
+                .addToFilters(createFilterDef(topic));
         try (var tester = kroxyliciousTester(config);
                 var producer = tester.producer()) {
             var invalid = producer.send(new ProducerRecord<>(topic.name(), "my-key", SYNTACTICALLY_INCORRECT_JSON));
@@ -52,7 +51,7 @@ class JsonSyntaxRecordValidationIT extends RecordValidationBaseIT {
         assertThat(cluster.getNumOfBrokers()).isOne();
 
         var config = proxy(cluster)
-                .addToFilters(createFilterDef(false, topic1));
+                .addToFilters(createFilterDef(topic1));
         try (var tester = kroxyliciousTester(config);
                 var producer = tester.producer()) {
             var rejected = producer.send(new ProducerRecord<>(topic1.name(), "my-key", SYNTACTICALLY_INCORRECT_JSON));
@@ -70,11 +69,11 @@ class JsonSyntaxRecordValidationIT extends RecordValidationBaseIT {
     }
 
     @Test
-    void partiallyInvalidJsonTransactionalAllRejected(KafkaCluster cluster, Topic topic1, Topic topic2) {
+    void invalidJsonProduceRejectedUsingTransaction(KafkaCluster cluster, Topic topic1, Topic topic2) {
         assertThat(cluster.getNumOfBrokers()).isOne();
 
         var config = proxy(cluster)
-                .addToFilters(createFilterDef(true, topic1));
+                .addToFilters(createFilterDef(topic1));
 
         try (var tester = kroxyliciousTester(config);
                 var producer = tester.producer(Map.of(LINGER_MS_CONFIG, 5000, TRANSACTIONAL_ID_CONFIG, randomUUID().toString()))) {
@@ -90,11 +89,11 @@ class JsonSyntaxRecordValidationIT extends RecordValidationBaseIT {
     }
 
     @Test
-    void partiallyInvalidJsonNotConfiguredToForwardAllRejected(KafkaCluster cluster, Topic topic1, Topic topic2) {
+    void singleValidationFailureCausesRejectionOfWholeBatch(KafkaCluster cluster, Topic topic1, Topic topic2) {
         assertThat(cluster.getNumOfBrokers()).isOne();
 
         var config = proxy(cluster)
-                .addToFilters(createFilterDef(false, topic1));
+                .addToFilters(createFilterDef(topic1));
 
         try (var tester = kroxyliciousTester(config);
                 var producer = tester.producer(Map.of(LINGER_MS_CONFIG, 5000))) {
@@ -107,35 +106,11 @@ class JsonSyntaxRecordValidationIT extends RecordValidationBaseIT {
     }
 
     @Test
-    void partiallyInvalidJsonProduceRejected(KafkaCluster cluster, Topic topic1, Topic topic2) {
+    void singleValidationFailureCausesRejectionOfWholeBatchSameTopic(KafkaCluster cluster, @TopicPartitions(2) Topic topic) {
         assertThat(cluster.getNumOfBrokers()).isOne();
 
         var config = proxy(cluster)
-                .addToFilters(createFilterDef(true, topic1));
-
-        try (var tester = kroxyliciousTester(config);
-                var producer = tester.producer(Map.of(LINGER_MS_CONFIG, 5000))) {
-            var invalid = producer.send(new ProducerRecord<>(topic1.name(), "my-key", SYNTACTICALLY_INCORRECT_JSON));
-            var valid = producer.send(new ProducerRecord<>(topic2.name(), "my-key", SYNTACTICALLY_CORRECT_JSON));
-            producer.flush();
-            assertThatFutureFails(invalid, InvalidRecordException.class, "value was not syntactically correct JSON");
-
-            assertThatFutureSucceeds(valid);
-
-            var records = consumeAll(tester, topic2);
-            assertThat(records)
-                    .singleElement()
-                    .extracting(ConsumerRecord::value)
-                    .isEqualTo(SYNTACTICALLY_CORRECT_JSON);
-        }
-    }
-
-    @Test
-    void partiallyInvalidAcrossPartitionsOfSameTopic(KafkaCluster cluster, @TopicPartitions(2) Topic topic) {
-        assertThat(cluster.getNumOfBrokers()).isOne();
-
-        var config = proxy(cluster)
-                .addToFilters(createFilterDef(true, topic));
+                .addToFilters(createFilterDef(topic));
 
         try (var tester = kroxyliciousTester(config);
                 var producer = tester.producer(Map.of(LINGER_MS_CONFIG, 5000))) {
@@ -143,39 +118,14 @@ class JsonSyntaxRecordValidationIT extends RecordValidationBaseIT {
             var valid = producer.send(new ProducerRecord<>(topic.name(), 1, "my-key", SYNTACTICALLY_CORRECT_JSON));
             producer.flush();
             assertThatFutureFails(invalid, InvalidRecordException.class, "value was not syntactically correct JSON");
-
-            assertThatFutureSucceeds(valid);
-
-            var records = consumeAll(tester, topic);
-            assertThat(records)
-                    .singleElement()
-                    .extracting(ConsumerRecord::value)
-                    .isEqualTo(SYNTACTICALLY_CORRECT_JSON);
-        }
-    }
-
-    @Test
-    void partiallyInvalidWithinOnePartitionOfTopic(KafkaCluster cluster, Topic topic) {
-        assertThat(cluster.getNumOfBrokers()).isOne();
-
-        var config = proxy(cluster)
-                .addToFilters(createFilterDef(true, topic));
-
-        try (var tester = kroxyliciousTester(config);
-                var producer = tester.producer(Map.of(LINGER_MS_CONFIG, 5000))) {
-            var invalid = producer.send(new ProducerRecord<>(topic.name(), "my-key", SYNTACTICALLY_INCORRECT_JSON));
-            var valid = producer.send(new ProducerRecord<>(topic.name(), "my-key", SYNTACTICALLY_CORRECT_JSON));
-            producer.flush();
-
-            assertThatFutureFails(invalid, InvalidRecordException.class, "value was not syntactically correct JSON");
-            assertThatFutureFails(valid, KafkaException.class, "Failed to append record because it was part of a batch which had one more more invalid records");
+            assertThatFutureFails(valid, InvalidRecordException.class, "Invalid record in another topic-partition caused whole ProduceRequest to be invalidated");
         }
     }
 
     @Test
     void validJsonProduceAccepted(KafkaCluster cluster, Topic topic) {
         var config = proxy(cluster)
-                .addToFilters(createFilterDef(false, topic));
+                .addToFilters(createFilterDef(topic));
 
         try (var tester = kroxyliciousTester(config);
                 var producer = tester.producer()) {
@@ -227,8 +177,8 @@ class JsonSyntaxRecordValidationIT extends RecordValidationBaseIT {
         }
     }
 
-    private FilterDefinition createFilterDef(boolean forwardPartialRequests, Topic... topics) {
-        return new FilterDefinitionBuilder(RecordValidation.class.getName()).withConfig("forwardPartialRequests", forwardPartialRequests, "rules",
+    private FilterDefinition createFilterDef(Topic... topics) {
+        return new FilterDefinitionBuilder(RecordValidation.class.getName()).withConfig("rules",
                 List.of(Map.of("topicNames", Arrays.stream(topics).map(Topic::name).toList(), "valueRule",
                         Map.of("syntacticallyCorrectJson", Map.of()))))
                 .build();