From b867176d6a92f5f3fab81cd0ff8670c3f95ccd63 Mon Sep 17 00:00:00 2001 From: gliljas Date: Wed, 4 Nov 2020 21:20:15 +0100 Subject: [PATCH 1/4] Add ClientId configuration --- .../Trigger/KafkaListenerConfiguration.cs | 7 +++++++ .../Trigger/KafkaTriggerAttribute.cs | 6 ++++++ .../Trigger/KafkaTriggerAttributeBindingProvider.cs | 1 + 3 files changed, 14 insertions(+) diff --git a/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Trigger/KafkaListenerConfiguration.cs b/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Trigger/KafkaListenerConfiguration.cs index 63e69cfe..7851519e 100644 --- a/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Trigger/KafkaListenerConfiguration.cs +++ b/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Trigger/KafkaListenerConfiguration.cs @@ -94,6 +94,13 @@ public class KafkaListenerConfiguration /// public string SslKeyPassword { get; set; } + /// + /// Client identifier. + /// Default: the hostname of the client machine + /// client.id in librdkafka + /// + public string ClientId { get; set; } + internal void ApplyToConfig(ClientConfig conf) { if (this.SaslMechanism.HasValue) diff --git a/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Trigger/KafkaTriggerAttribute.cs b/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Trigger/KafkaTriggerAttribute.cs index e8cb85cf..56f7b074 100644 --- a/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Trigger/KafkaTriggerAttribute.cs +++ b/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Trigger/KafkaTriggerAttribute.cs @@ -107,6 +107,12 @@ public KafkaTriggerAttribute(string brokerList, string topic) /// public string SslKeyPassword { get; set; } + /// + /// Client identifier. + /// Default: the hostname of the client machine + /// client.id in librdkafka + /// + public string ClientId { get; set; } bool IsValidValueType(Type value) { diff --git a/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Trigger/KafkaTriggerAttributeBindingProvider.cs b/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Trigger/KafkaTriggerAttributeBindingProvider.cs index 22218234..b8f27104 100644 --- a/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Trigger/KafkaTriggerAttributeBindingProvider.cs +++ b/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Trigger/KafkaTriggerAttributeBindingProvider.cs @@ -92,6 +92,7 @@ private KafkaListenerConfiguration CreateConsumerConfiguration(KafkaTriggerAttri ConsumerGroup = this.config.ResolveSecureSetting(nameResolver, attribute.ConsumerGroup), Topic = this.config.ResolveSecureSetting(nameResolver, attribute.Topic), EventHubConnectionString = this.config.ResolveSecureSetting(nameResolver, attribute.EventHubConnectionString), + ClientId = this.config.ResolveSecureSetting(nameResolver, attribute.ClientId) }; if (attribute.AuthenticationMode != BrokerAuthenticationMode.NotSet || From 4050850ee0dfa50f6124bb63def111a785a48159 Mon Sep 17 00:00:00 2001 From: gliljas Date: Wed, 4 Nov 2020 21:40:57 +0100 Subject: [PATCH 2/4] Set clientid on producerconfig --- .../Output/KafkaAttribute.cs | 7 +++ .../Output/KafkaProducerFactory.cs | 4 +- .../KafkaProducerFactoryTest.cs | 50 +++++++++++++++++++ 3 files changed, 60 insertions(+), 1 deletion(-) diff --git a/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Output/KafkaAttribute.cs b/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Output/KafkaAttribute.cs index 74cc80b0..1747b376 100644 --- a/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Output/KafkaAttribute.cs +++ b/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Output/KafkaAttribute.cs @@ -139,5 +139,12 @@ public KafkaAttribute() /// ssl.key.password in librdkafka /// public string SslKeyPassword { get; set; } + + /// + /// Client identifier. + /// Default: the hostname of the client machine + /// client.id in librdkafka + /// + public string ClientId { get; set; } } } \ No newline at end of file diff --git a/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Output/KafkaProducerFactory.cs b/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Output/KafkaProducerFactory.cs index 178f3ffc..3a8280ce 100644 --- a/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Output/KafkaProducerFactory.cs +++ b/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Output/KafkaProducerFactory.cs @@ -3,6 +3,7 @@ using System; using System.Collections.Concurrent; +using System.Net; using System.Reflection; using System.Text; using Avro.Generic; @@ -130,7 +131,8 @@ public ProducerConfig GetProducerConfig(KafkaProducerEntity entity) SslCaLocation = resolvedSslCaLocation, Debug = kafkaOptions?.LibkafkaDebug, MetadataMaxAgeMs = kafkaOptions?.MetadataMaxAgeMs, - SocketKeepaliveEnable = kafkaOptions?.SocketKeepaliveEnable + SocketKeepaliveEnable = kafkaOptions?.SocketKeepaliveEnable, + ClientId = this.config.ResolveSecureSetting(nameResolver, entity.Attribute.ClientId) ?? Dns.GetHostName() }; if (entity.Attribute.AuthenticationMode != BrokerAuthenticationMode.NotSet) diff --git a/test/Microsoft.Azure.WebJobs.Extensions.Kafka.UnitTests/KafkaProducerFactoryTest.cs b/test/Microsoft.Azure.WebJobs.Extensions.Kafka.UnitTests/KafkaProducerFactoryTest.cs index a0221620..f93f5033 100644 --- a/test/Microsoft.Azure.WebJobs.Extensions.Kafka.UnitTests/KafkaProducerFactoryTest.cs +++ b/test/Microsoft.Azure.WebJobs.Extensions.Kafka.UnitTests/KafkaProducerFactoryTest.cs @@ -5,6 +5,7 @@ using System.Collections.Generic; using System.IO; using System.Linq; +using System.Net; using System.Reflection; using Avro.Generic; @@ -304,5 +305,54 @@ public void GetProducerConfig_When_Ssl_Locations_Resolve_InAzure_Should_Contain_ Assert.Equal(sslCa.FullName, config.SslCaLocation); Assert.Equal(sslKeyLocation.FullName, config.SslKeyLocation); } + + [Fact] + public void GetProducerConfig_When_Client_Defined_Should_SetFromConfig() + { + var attribute = new KafkaAttribute("brokers:9092", "myTopic") + { + AuthenticationMode = BrokerAuthenticationMode.Plain, + Protocol = BrokerProtocol.SaslSsl, + Username = "myuser", + Password = "secret", + ClientId = "%clientid%" + }; + + var entity = new KafkaProducerEntity() + { + Attribute = attribute, + ValueType = typeof(ProtoUser), + }; + + var configuration = new ConfigurationBuilder().AddInMemoryCollection(new Dictionary { + ["clientid"] = "SpecifiedClientId" + }).Build(); + + var factory = new KafkaProducerFactory(configuration, new DefaultNameResolver(configuration), NullLoggerProvider.Instance); + var config = factory.GetProducerConfig(entity); + Assert.Equal("SpecifiedClientId", config.ClientId); + } + + [Fact] + public void GetProducerConfig_When_Client_Not_Defined_Should_SetHostName() + { + var attribute = new KafkaAttribute("brokers:9092", "myTopic") + { + AuthenticationMode = BrokerAuthenticationMode.Plain, + Protocol = BrokerProtocol.SaslSsl, + Username = "myuser", + Password = "secret" + }; + + var entity = new KafkaProducerEntity() + { + Attribute = attribute, + ValueType = typeof(ProtoUser), + }; + + var factory = new KafkaProducerFactory(emptyConfiguration, new DefaultNameResolver(emptyConfiguration), NullLoggerProvider.Instance); + var config = factory.GetProducerConfig(entity); + Assert.Equal(Dns.GetHostName(), config.ClientId); + } } } \ No newline at end of file From 8823bc742f9654663d48e5b341509669e2106bec Mon Sep 17 00:00:00 2001 From: gliljas Date: Wed, 4 Nov 2020 21:42:25 +0100 Subject: [PATCH 3/4] Documentation --- README.md | 1 + 1 file changed, 1 insertion(+) diff --git a/README.md b/README.md index b0be6c98..9859afc1 100644 --- a/README.md +++ b/README.md @@ -322,6 +322,7 @@ The settings exposed here are targeted to more advanced users that want to custo |LibkafkaDebug|debug|Both |MetadataMaxAgeMs|metadata.max.age.ms|Both |SocketKeepaliveEnable|socket.keepalive.enable|Both +|ClientId|client.id|Both **NOTE:** `MetadataMaxAgeMs` default is `180000` `SocketKeepaliveEnable` default is `true` otherwise, the default value is the same as the [Configuration properties](https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md). The reason of the default settings, refer to this [issue](https://github.com/Azure/azure-functions-kafka-extension/issues/187). From 8e20830279b6cf6d4ee2e0ea5d289839900ce228 Mon Sep 17 00:00:00 2001 From: gliljas Date: Thu, 5 Nov 2020 15:49:10 +0100 Subject: [PATCH 4/4] Test and missing code --- .../Listeners/KafkaListener.cs | 3 ++ .../KafkaListenerTest.cs | 40 ++++++++++++++++++- 2 files changed, 41 insertions(+), 2 deletions(-) diff --git a/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Listeners/KafkaListener.cs b/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Listeners/KafkaListener.cs index 3511fd36..af6a859a 100644 --- a/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Listeners/KafkaListener.cs +++ b/src/Microsoft.Azure.WebJobs.Extensions.Kafka/Listeners/KafkaListener.cs @@ -4,6 +4,7 @@ using System; using System.Collections.Generic; using System.Linq; +using System.Net; using System.Threading; using System.Threading.Tasks; using Confluent.Kafka; @@ -166,6 +167,8 @@ private ConsumerConfig GetConsumerConfiguration() // start from earliest if no checkpoint has been committed AutoOffsetReset = AutoOffsetReset.Earliest, + ClientId = this.listenerConfiguration.ClientId ?? Dns.GetHostName(), + // Secure communication/authentication SaslMechanism = this.listenerConfiguration.SaslMechanism, SaslUsername = this.listenerConfiguration.SaslUsername, diff --git a/test/Microsoft.Azure.WebJobs.Extensions.Kafka.UnitTests/KafkaListenerTest.cs b/test/Microsoft.Azure.WebJobs.Extensions.Kafka.UnitTests/KafkaListenerTest.cs index 804f47d7..e492a014 100644 --- a/test/Microsoft.Azure.WebJobs.Extensions.Kafka.UnitTests/KafkaListenerTest.cs +++ b/test/Microsoft.Azure.WebJobs.Extensions.Kafka.UnitTests/KafkaListenerTest.cs @@ -10,6 +10,7 @@ using System.Collections.Concurrent; using System.IO; using System.Linq; +using System.Net; using System.Threading; using System.Threading.Tasks; using Xunit; @@ -342,7 +343,7 @@ public async Task When_Options_Are_Set_Should_Be_Set_In_Consumer_Config() await target.StartAsync(default); - Assert.Equal(12, target.ConsumerConfig.Count()); + Assert.Equal(13, target.ConsumerConfig.Count()); Assert.Equal("testBroker", target.ConsumerConfig.BootstrapServers); Assert.Equal("group1", target.ConsumerConfig.GroupId); Assert.Equal("password1", target.ConsumerConfig.SslKeyPassword); @@ -355,10 +356,45 @@ public async Task When_Options_Are_Set_Should_Be_Set_In_Consumer_Config() Assert.Equal(180000, target.ConsumerConfig.MetadataMaxAgeMs); Assert.Equal(true, target.ConsumerConfig.SocketKeepaliveEnable); Assert.Equal(AutoOffsetReset.Earliest, target.ConsumerConfig.AutoOffsetReset); + Assert.Equal(Dns.GetHostName(), target.ConsumerConfig.ClientId); await target.StopAsync(default); } + [Fact] + public async Task When_ClientId_Is_Set_Should_Be_Set_In_Consumer_Config() + { + var executor = new Mock(); + var consumer = new Mock>(); + + var listenerConfig = new KafkaListenerConfiguration() + { + BrokerList = "testBroker", + Topic = "topic", + ClientId = "testclientid" + }; + + var kafkaOptions = new KafkaOptions(); + var target = new KafkaListenerForTest( + executor.Object, + true, + kafkaOptions, + listenerConfig, + requiresKey: true, + valueDeserializer: null, + NullLogger.Instance, + functionId: "testId" + ); + + target.SetConsumer(consumer.Object); + + await target.StartAsync(default); + + Assert.Equal("testclientid", target.ConsumerConfig.ClientId); + + await target.StopAsync(default); + } + [Fact] public async Task When_Options_With_Ssal_Are_Set_Should_Be_Set_In_Consumer_Config() { @@ -392,7 +428,7 @@ public async Task When_Options_With_Ssal_Are_Set_Should_Be_Set_In_Consumer_Confi await target.StartAsync(default); - Assert.Equal(12, target.ConsumerConfig.Count()); + Assert.Equal(13, target.ConsumerConfig.Count()); Assert.Equal("testBroker", target.ConsumerConfig.BootstrapServers); Assert.Equal("group1", target.ConsumerConfig.GroupId); Assert.Equal(kafkaOptions.AutoCommitIntervalMs, target.ConsumerConfig.AutoCommitIntervalMs);