Browse Source
Merge pull request #13907 from abpframework/liangshiwei/kafka
Fix Kafka connection problem
pull/13909/head
maliming
4 years ago
committed by
GitHub
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with
2 additions and
2 deletions
-
framework/src/Volo.Abp.Kafka/Volo/Abp/Kafka/ConsumerPool.cs
-
framework/src/Volo.Abp.Kafka/Volo/Abp/Kafka/ProducerPool.cs
|
|
|
@ -37,7 +37,7 @@ public class ConsumerPool : IConsumerPool, ISingletonDependency |
|
|
|
return Consumers.GetOrAdd( |
|
|
|
connectionName, connection => new Lazy<IConsumer<string, byte[]>>(() => |
|
|
|
{ |
|
|
|
var config = new ConsumerConfig(Options.Connections.GetOrDefault(connection)) |
|
|
|
var config = new ConsumerConfig(Options.Connections.GetOrDefault(connection).ToDictionary(k => k.Key, v => v.Value)) |
|
|
|
{ |
|
|
|
GroupId = groupId, |
|
|
|
EnableAutoCommit = false |
|
|
|
|
|
|
|
@ -39,7 +39,7 @@ public class ProducerPool : IProducerPool, ISingletonDependency |
|
|
|
return Producers.GetOrAdd( |
|
|
|
connectionName, connection => new Lazy<IProducer<string, byte[]>>(() => |
|
|
|
{ |
|
|
|
var producerConfig = new ProducerConfig(Options.Connections.GetOrDefault(connection)); |
|
|
|
var producerConfig = new ProducerConfig(Options.Connections.GetOrDefault(connection).ToDictionary(k => k.Key, v => v.Value)); |
|
|
|
Options.ConfigureProducer?.Invoke(producerConfig); |
|
|
|
return new ProducerBuilder<string, byte[]>(producerConfig).Build(); |
|
|
|
|
|
|
|
|