Discard invalid consumer records

Signed-off-by: Pierangelo Di Pilato <[email protected]>
knative-extensions · Oct 16, 2024 · 57ad189 · 57ad189
1 parent d1e6cb4
commit 57ad189
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 2 deletions.
diff --git a/.../java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherMutatorChain.java b/.../java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherMutatorChain.java
@@ -38,8 +38,11 @@ public RecordDispatcherMutatorChain(final RecordDispatcher next, final CloudEven
 
     @Override
     public Future<Void> dispatch(ConsumerRecord<Object, CloudEvent> record) {
-        return next.dispatch(
-                KafkaConsumerRecordUtils.copyRecordAssigningValue(record, cloudEventMutator.apply(record)));
+        final var n = cloudEventMutator.apply(record);
+        if (n == record.value()) {
+            return next.dispatch(record);
+        }
+        return next.dispatch(KafkaConsumerRecordUtils.copyRecordAssigningValue(record, n));
     }
 
     @Override

diff --git a/...ev/knative/eventing/kafka/broker/dispatcher/impl/consumer/CloudEventOverridesMutator.java b/...ev/knative/eventing/kafka/broker/dispatcher/impl/consumer/CloudEventOverridesMutator.java
@@ -35,6 +35,9 @@ public CloudEventOverridesMutator(final DataPlaneContract.CloudEventOverrides cl
 
     @Override
     public CloudEvent apply(ConsumerRecord<Object, CloudEvent> record) {
+        if (record.value() instanceof InvalidCloudEvent) {
+            return record.value();
+        }
         final var builder = CloudEventBuilder.from(record.value());
         applyKafkaMetadata(builder, record.partition(), record.offset());
         applyCloudEventOverrides(builder);