Regenerate configs upon consumer/producer creation

RJSonnenberg · Apr 21, 2023 · d942b67 · d942b67
1 parent 863d71d
commit d942b67
Show file tree

Hide file tree

Showing 2 changed files with 5 additions and 5 deletions.
diff --git a/...MassTransit.KafkaIntegration/KafkaIntegration/Configuration/KafkaProducerSpecification.cs b/...MassTransit.KafkaIntegration/KafkaIntegration/Configuration/KafkaProducerSpecification.cs
@@ -159,10 +159,10 @@ public void SetHeadersSerializer(IHeadersSerializer serializer)
 
         public KafkaSendTransportContext<TKey, TValue> CreateSendTransportContext(IBusInstance busInstance, Action onStop = null)
         {
-            var producerConfig = _hostConfiguration.GetProducerConfig(_producerConfig);
-
             ProducerBuilder<byte[], byte[]> CreateProducerBuilder()
             {
+                var producerConfig = _hostConfiguration.GetProducerConfig(_producerConfig);
+
                 ProducerBuilder<byte[], byte[]> producerBuilder = new ProducerBuilder<byte[], byte[]>(producerConfig)
                     .SetKeySerializer(Serializers.ByteArray)
                     .SetValueSerializer(Serializers.ByteArray);

diff --git a/...KafkaIntegration/KafkaIntegration/Configuration/KafkaTopicReceiveEndpointConfiguration.cs b/...KafkaIntegration/KafkaIntegration/Configuration/KafkaTopicReceiveEndpointConfiguration.cs
@@ -205,10 +205,10 @@ public override ReceiveEndpointContext CreateReceiveEndpointContext()
 
         KafkaReceiveEndpointContext<TKey, TValue> CreateReceiveKafkaEndpointContext()
         {
-            var consumerConfig = _hostConfiguration.GetConsumerConfig(_consumerConfig);
-
             ConsumerBuilder<byte[], byte[]> CreateConsumerBuilder()
             {
+                var consumerConfig = _hostConfiguration.GetConsumerConfig(_consumerConfig);
+
                 ConsumerBuilder<byte[], byte[]> consumerBuilder = new ConsumerBuilder<byte[], byte[]>(consumerConfig)
                     .SetLogHandler((c, message) => _busInstance.HostConfiguration.ReceiveLogContext?.Debug?.Log(message.Message));
 
@@ -222,7 +222,7 @@ ConsumerBuilder<byte[], byte[]> CreateConsumerBuilder()
                 return consumerBuilder;
             }
 
-            var builder = new KafkaReceiveEndpointBuilder<TKey, TValue>(_busInstance, _hostConfiguration, consumerConfig.GroupId, this,
+            var builder = new KafkaReceiveEndpointBuilder<TKey, TValue>(_busInstance, _hostConfiguration, _consumerConfig.GroupId, this,
                 this, _headersDeserializer, _keyDeserializer, _valueDeserializer, CreateConsumerBuilder);
             ApplySpecifications(builder);