Merge branch 'main' into sslTimeouts

Author: SamBarker

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
 

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.
 

kroxylicious-filters/kroxylicious-record-validation/src/main/java/io/kroxylicious/proxy/filter/validation/ProduceRequestValidatorBuilder.java

@@ -38,8 +38,8 @@ private 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();

kroxylicious-filters/kroxylicious-record-validation/src/main/java/io/kroxylicious/proxy/filter/validation/RecordValidation.java

@@ -24,7 +24,7 @@ public ValidationConfig initialize(FilterFactoryContext context, ValidationConfi
     @Override
     public RecordValidationFilter createFilter(FilterFactoryContext context, ValidationConfig configuration) {
         ProduceRequestValidator validator = ProduceRequestValidatorBuilder.build(configuration);
-        return new RecordValidationFilter(configuration.isForwardPartialRequests(), validator);
+        return new RecordValidationFilter(validator);
     }
 
 }

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.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,37 +58,19 @@ public CompletionStage<RequestFilterResult> onProduceRequest(short apiVersion, R
         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);
             }
         });
     }
 
-    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) {

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 +
                 '}';
     }

kroxylicious-filters/kroxylicious-record-validation/src/main/java/io/kroxylicious/proxy/filter/validation/validators/request/ProduceRequestValidationResult.java

@@ -9,7 +9,6 @@
 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 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

kroxylicious-filters/kroxylicious-record-validation/src/main/java/io/kroxylicious/proxy/filter/validation/validators/topic/AllValidTopicValidationResult.java

@@ -16,11 +16,6 @@ public boolean isAnyPartitionInvalid() {
         return false;
     }
 
-    @Override
-    public boolean isAllPartitionsInvalid() {
-        return false;
-    }
-
     @Override
     public Stream<PartitionValidationResult> invalidPartitions() {
         return Stream.empty();

kroxylicious-filters/kroxylicious-record-validation/src/main/java/io/kroxylicious/proxy/filter/validation/validators/topic/PerPartitionTopicValidationResult.java

@@ -16,14 +16,6 @@ public boolean isAnyPartitionInvalid() {
         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());

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