From e078cd5e543581b0c771201838d4bac68eb2cfef Mon Sep 17 00:00:00 2001
From: Chris Patterson <chris@phatboyg.com>
Date: Tue, 2 May 2023 15:48:18 -0500
Subject: [PATCH] Fixed #4345 - discard skipped/faulted messages for
 Kafka/Event Hub

---
 .../Configuration/EventHubReceiveEndpointConfigurator.cs       | 3 +++
 .../Configuration/KafkaTopicReceiveEndpointConfiguration.cs    | 3 +++
 2 files changed, 6 insertions(+)

diff --git a/src/Transports/MassTransit.EventHubIntegration/EventHubIntegration/Configuration/EventHubReceiveEndpointConfigurator.cs b/src/Transports/MassTransit.EventHubIntegration/EventHubIntegration/Configuration/EventHubReceiveEndpointConfigurator.cs
index 9a8f807d5d8..bdbe8bb408c 100644
--- a/src/Transports/MassTransit.EventHubIntegration/EventHubIntegration/Configuration/EventHubReceiveEndpointConfigurator.cs
+++ b/src/Transports/MassTransit.EventHubIntegration/EventHubIntegration/Configuration/EventHubReceiveEndpointConfigurator.cs
@@ -55,6 +55,9 @@ public EventHubReceiveEndpointConfigurator(IEventHubHostConfiguration hostConfig
             _blobClient = new Lazy<BlobContainerClient>(CreateBlobClient);
 
             PublishFaults = false;
+
+            this.DiscardFaultedMessages();
+            this.DiscardSkippedMessages();
         }
 
         public override Uri HostAddress => _endpointConfiguration.HostAddress;
diff --git a/src/Transports/MassTransit.KafkaIntegration/KafkaIntegration/Configuration/KafkaTopicReceiveEndpointConfiguration.cs b/src/Transports/MassTransit.KafkaIntegration/KafkaIntegration/Configuration/KafkaTopicReceiveEndpointConfiguration.cs
index 7e031355914..c53f39da40a 100644
--- a/src/Transports/MassTransit.KafkaIntegration/KafkaIntegration/Configuration/KafkaTopicReceiveEndpointConfiguration.cs
+++ b/src/Transports/MassTransit.KafkaIntegration/KafkaIntegration/Configuration/KafkaTopicReceiveEndpointConfiguration.cs
@@ -55,6 +55,9 @@ public KafkaTopicReceiveEndpointConfiguration(IKafkaHostConfiguration hostConfig
             _consumerConfigurator = new PipeConfigurator<ConsumerContext>();
 
             PublishFaults = false;
+
+            this.DiscardFaultedMessages();
+            this.DiscardSkippedMessages();
         }
 
         public override Uri HostAddress => _endpointConfiguration.HostAddress;