Logging for schema registry client creation #41

.editorconfig

@@ -62,6 +62,7 @@ dotnet_diagnostic.CA1507.severity = warning
 dotnet_diagnostic.CA1805.severity = warning
 csharp_style_prefer_primary_constructors = false:suggestion
 csharp_style_prefer_readonly_struct_member = true:suggestion
+csharp_prefer_system_threading_lock = true:suggestion
 [*.{cs,vb}]
 #### Naming styles ####
 
@@ -143,6 +144,9 @@ dotnet_style_qualification_for_field = true:suggestion
 dotnet_style_qualification_for_property = true:suggestion
 dotnet_style_qualification_for_method = true:suggestion
 dotnet_style_qualification_for_event = true:suggestion
+csharp_style_unused_value_expression_statement_preference = discard_variable:none
+dotnet_diagnostic.IDE0055.severity = none
+dotnet_diagnostic.IDE0057.severity = none
 dotnet_diagnostic.CA1051.severity = warning
 dotnet_diagnostic.CA1068.severity = warning
 dotnet_diagnostic.CA1069.severity = error
@@ -178,7 +182,7 @@ dotnet_diagnostic.CA3075.severity = warning
 dotnet_diagnostic.CA2201.severity = warning
 dotnet_diagnostic.CA2219.severity = warning
 dotnet_diagnostic.CA2251.severity = suggestion
-dotnet_style_prefer_collection_expression = false:warning
+dotnet_style_prefer_collection_expression = never:warning
 dotnet_diagnostic.CA1510.severity = none
 dotnet_diagnostic.CA1511.severity = none
 dotnet_diagnostic.CA1512.severity = none

.github/workflows/dotnet.yml

@@ -32,7 +32,7 @@ jobs:
     - name: Setup .NET
       uses: actions/setup-dotnet@v4
       with:
-        dotnet-version: 8.0.x
+        dotnet-version: 9.0.x
     - name: Restore
       run: dotnet restore
     - name: Build
@@ -85,7 +85,7 @@ jobs:
     - name: Setup .NET
       uses: actions/setup-dotnet@v4
       with:
-        dotnet-version: 8.0.x
+        dotnet-version: 9.0.x
         source-url: https://api.nuget.org/v3/index.json
       env:
         NUGET_AUTH_TOKEN: ${{secrets.NUGET_TOKEN}}

sample/Epam.Kafka.Sample/Epam.Kafka.Sample.csproj

@@ -20,10 +20,10 @@
   </ItemGroup>
 
   <ItemGroup>
-    <PackageReference Include="Microsoft.EntityFrameworkCore.InMemory" Version="8.0.1" />
-    <PackageReference Include="Microsoft.Extensions.Hosting" Version="8.0.0" />
-    <PackageReference Include="OpenTelemetry.Exporter.Console" Version="1.7.0" />
-    <PackageReference Include="OpenTelemetry.Extensions.Hosting" Version="1.7.0" />
+    <PackageReference Include="Microsoft.EntityFrameworkCore.InMemory" Version="8.0.11" />
+    <PackageReference Include="Microsoft.Extensions.Hosting" Version="8.0.1" />
+    <PackageReference Include="OpenTelemetry.Exporter.Console" Version="1.10.0" />
+    <PackageReference Include="OpenTelemetry.Extensions.Hosting" Version="1.10.0" />
 
   </ItemGroup>
 

src/Directory.Build.props

@@ -25,7 +25,7 @@
 		<NeutralLanguage>en</NeutralLanguage>
 		<PackageReleaseNotes>https://github.com/epam/epam-kafka/releases</PackageReleaseNotes>
 		<VersionPrefix>0</VersionPrefix>
-		<Version>2.3.$(VersionPrefix)</Version>
+		<Version>2.4.$(VersionPrefix)</Version>
 	</PropertyGroup>
 
 	<ItemGroup>

src/Epam.Kafka.PubSub.EntityFramework6/CompatibilityExtensions.cs

@@ -13,7 +13,7 @@ public static IQueryable<T> AsTracking<T>(this IQueryable<T> queryable)
         return queryable;
     }
 
-    public static int SaveChanges(this DbContext context, bool acceptAllChangesOnSuccess)
+    public static int SaveChanges(this DbContext context, bool _)
     {
         return context.SaveChanges();
     }

src/Epam.Kafka.PubSub.EntityFrameworkCore/Epam.Kafka.PubSub.EntityFrameworkCore.csproj

@@ -10,11 +10,11 @@
 	</ItemGroup>
 
 	<ItemGroup Condition="'$(TargetFramework)' == 'net6.0'">
-		<PackageReference Include="Microsoft.EntityFrameworkCore" Version="6.0.0" />
+		<PackageReference Include="Microsoft.EntityFrameworkCore" Version="6.0.36" />
 	</ItemGroup>
 
 	<ItemGroup Condition="'$(TargetFramework)' == 'net8.0'">
-		<PackageReference Include="Microsoft.EntityFrameworkCore" Version="8.0.0" />
+		<PackageReference Include="Microsoft.EntityFrameworkCore" Version="8.0.11" />
 	</ItemGroup>
 
 	<ItemGroup>

src/Epam.Kafka.PubSub.EntityFrameworkCore/Publication/Contracts/DbContextEntityPublicationHandler.cs

@@ -84,10 +84,12 @@ protected override void TransactionCommitted(IReadOnlyCollection<TEntity> entiti
         }
         catch (DbUpdateConcurrencyException exception)
         {
+#pragma warning disable IDE0008 // Use explicit type not possible due to #if directives
             foreach (var entry in exception.Entries)
             {
                 this.Logger.PublicationEntityDetached(exception, "Commit", this.FindPrimaryKeyForLogs(entry), typeof(TEntity));
             }
+#pragma warning restore IDE0008
         }
     }
 

src/Epam.Kafka.PubSub.EntityFrameworkCore/Subscription/DbContextOffsetsStorage.cs

@@ -127,6 +127,7 @@ public IReadOnlyCollection<TopicPartitionOffset> CommitOrReset(
         {
             cancellationToken.ThrowIfCancellationRequested();
 
+#pragma warning disable IDE0008 // Use explicit type not possible due to #if directives
             foreach (var entry in exception.Entries)
             {
                 if (entry.Entity is KafkaTopicState)
@@ -152,6 +153,7 @@ public IReadOnlyCollection<TopicPartitionOffset> CommitOrReset(
                     }
                 }
             }
+#pragma warning restore IDE0008
 
             this._context.SaveChanges(true);
         }

src/Epam.Kafka.PubSub.EntityFrameworkCore/Subscription/State/KafkaTopicStateExtensions.cs

@@ -1,5 +1,8 @@
 // Copyright © 2024 EPAM Systems
 
+#pragma warning disable IDE0065
+#pragma warning disable IDE0008
+
 #if EF6
 namespace Epam.Kafka.PubSub.EntityFramework6.Subscription.State;
 #else
@@ -45,4 +48,7 @@ public static ETBuilder AddKafkaState(this ModelBuilder builder)
 
         return e;
     }
-}
+}
+
+#pragma warning restore IDE0065
+#pragma warning restore IDE0008

src/Epam.Kafka.PubSub/Common/HealthChecks/PubSubHealthCheck.cs

@@ -52,7 +52,13 @@ protected PubSubHealthCheck(PubSubOptions options)
             default:
                 throw new InvalidOperationException(
                     $"Unknown pipeline status {monitor.Pipeline.Value}. Supported values: " +
-                    string.Join(", ", Enum.GetNames(typeof(PipelineStatus))));
+                    string.Join(", ",
+#if NET6_0_OR_GREATER
+                        Enum.GetNames<PipelineStatus>()
+#else
+                        Enum.GetNames(typeof(PipelineStatus))
+#endif
+                        ));
         }
 
         return Task.FromResult(new HealthCheckResult(

src/Epam.Kafka.PubSub/Common/PubSubBackgroundService.cs

@@ -59,7 +59,7 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
     {
         await Task.Yield();
 
-        if (this.Options.Enabled == false)
+        if (!this.Options.Enabled)
         {
             this.Monitor.Pipeline.Update(PipelineStatus.Disabled);
             return;

src/Epam.Kafka.PubSub/Publication/Options/PublicationOptionsValidate.cs

@@ -16,7 +16,7 @@ public ValidateOptionsResult Validate(string? name, PublicationOptions options)
             throw new ArgumentNullException(nameof(options));
         }
 
-        if (options.Enabled == false)
+        if (!options.Enabled)
         {
             return ValidateOptionsResult.Success;
         }

src/Epam.Kafka.PubSub/Subscription/Options/SubscriptionOptionsValidate.cs

@@ -17,7 +17,7 @@ public ValidateOptionsResult Validate(string? name, SubscriptionOptions options)
             throw new ArgumentNullException(nameof(options));
         }
 
-        if (options.Enabled == false)
+        if (!options.Enabled)
         {
             return ValidateOptionsResult.Success;
         }

src/Epam.Kafka.PubSub/Subscription/Pipeline/SubscriptionBatchResult.cs

@@ -1,7 +1,5 @@
 // Copyright © 2024 EPAM Systems
 
-using Confluent.Kafka;
-
 using Epam.Kafka.PubSub.Common.Pipeline;
 
 namespace Epam.Kafka.PubSub.Subscription.Pipeline;

src/Epam.Kafka.PubSub/Subscription/State/ExternalState.cs

@@ -40,7 +40,7 @@ protected override void AssignConsumer<TKey, TValue>(SubscriptionTopicWrapper<TK
             // existing assignment, check if offset reset
             if (topic.Consumer.Assignment.Contains(item.TopicPartition))
             {
-                if (topic.TryGetOffset(item.TopicPartition, out var tp) && tp != item.Offset)
+                if (topic.TryGetOffset(item.TopicPartition, out Offset tp) && tp != item.Offset)
                 {
                     ExternalStateExtensions.PauseOrReset(topic, item, pause, reset);
                 }

src/Epam.Kafka.PubSub/Subscription/State/ExternalStateExtensions.cs

@@ -62,8 +62,8 @@ public static IReadOnlyCollection<TopicPartitionOffset> CommitState<TKey, TValue
     }
 
     public static void PauseOrReset<TKey, TValue>(
-        SubscriptionTopicWrapper<TKey, TValue> topic, 
-        TopicPartitionOffset item, 
+        SubscriptionTopicWrapper<TKey, TValue> topic,
+        TopicPartitionOffset item,
         List<TopicPartition> pause,
         List<TopicPartitionOffset> reset)
     {
@@ -112,7 +112,7 @@ public static void CommitOffsetIfNeeded<TKey, TValue>(
                     }
                     else if (item.Offset == Offset.End)
                     {
-                        var w = topic.Consumer.QueryWatermarkOffsets(item.TopicPartition, TimeSpan.FromSeconds(5));
+                        WatermarkOffsets w = topic.Consumer.QueryWatermarkOffsets(item.TopicPartition, TimeSpan.FromSeconds(5));
                         toCommit.Add(new TopicPartitionOffset(item.TopicPartition, w.High));
                     }
                 }

src/Epam.Kafka.PubSub/Subscription/SubscriptionBackgroundService.cs

@@ -101,7 +101,7 @@ protected override void ExecuteBatch(
         this.Monitor.Batch.Update(BatchStatus.Reading);
 
         bool unassignedBeforeRead = state.GetBatch(
-            topic, activitySpan, out var batch, cancellationToken);
+            topic, activitySpan, out IReadOnlyCollection<ConsumeResult<TKey, TValue>> batch, cancellationToken);
 
         cancellationToken.ThrowIfCancellationRequested();
 

src/Epam.Kafka.PubSub/Subscription/Topics/SubscriptionTopicWrapper.cs

@@ -79,7 +79,10 @@ public SubscriptionTopicWrapper(IKafkaFactory kafkaFactory,
 
     public Func<IReadOnlyCollection<TopicPartition>, IReadOnlyCollection<TopicPartitionOffset>>? ExternalState { get; set; }
 
-    public bool TryGetOffset(TopicPartition tp, out Offset result) => this._offsets.TryGetValue(tp, out result);
+    public bool TryGetOffset(TopicPartition tp, out Offset result)
+    {
+        return this._offsets.TryGetValue(tp, out result);
+    }
 
     public void OnAssign(IReadOnlyCollection<TopicPartitionOffset> items)
     {
@@ -229,7 +232,7 @@ public void OnReset(IReadOnlyCollection<TopicPartitionOffset> items)
             List<TopicPartitionOffset> reset = new(items.Count);
             List<TopicPartitionOffset> resume = new(items.Count);
 
-            foreach (var tpo in items)
+            foreach (TopicPartitionOffset tpo in items)
             {
                 if (this._paused.Remove(tpo.TopicPartition))
                 {
@@ -285,7 +288,7 @@ private bool OnPauseEnumerate(IEnumerable<TopicPartition> items)
                 throw;
             }
 
-            foreach (var r in result)
+            foreach (TopicPartition r in result)
             {
                 this._paused.Add(r);
                 this._offsets[r] = ExternalOffset.Paused;
@@ -485,6 +488,7 @@ private void ReadToBuffer(ActivityWrapper span, CancellationToken cancellationTo
         {
             ConsumeResult<byte[], byte[]> record = consumeException.ConsumerRecord;
 
+#pragma warning disable IDE0010 // Add missing cases
             switch (consumeException.Error.Code)
             {
                 case ErrorCode.Local_Fatal:
@@ -507,6 +511,7 @@ private void ReadToBuffer(ActivityWrapper span, CancellationToken cancellationTo
                         throw;
                     }
             }
+#pragma warning restore IDE0010 // Add missing cases
 
             // unable to return at least something, so only throw
             if (this._buffer.Count == 0)

src/Epam.Kafka.PubSub/Utils/ActivityWrapper.cs

@@ -5,6 +5,7 @@
 using System.Text;
 
 namespace Epam.Kafka.PubSub.Utils;
+
 #pragma warning disable CA2000 // StartActivity returns same activity object, so it will be disposed.
 internal sealed class ActivityWrapper : IDisposable
 {

src/Epam.Kafka/Internals/KafkaFactory.cs

@@ -220,15 +220,16 @@ public IClient GetOrCreateClient(string? cluster = null)
 
         KafkaClusterOptions clusterOptions = this.GetAndValidateClusterOptions(cluster);
 
-        SharedClient? result;
-
-        lock (this._syncObj)
+        if (!this._clients.TryGetValue(clusterOptions, out SharedClient? result))
         {
-            if (!this._clients.TryGetValue(clusterOptions, out result))
+            lock (this._syncObj)
             {
-                result = new SharedClient(this, cluster);
+                if (!this._clients.TryGetValue(clusterOptions, out result))
+                {
+                    result = new SharedClient(this, cluster);
 
-                this._clients.Add(clusterOptions, result);
+                    this._clients.Add(clusterOptions, result);
+                }
             }
         }
 
@@ -241,16 +242,31 @@ public ISchemaRegistryClient GetOrCreateSchemaRegistryClient(string? cluster = n
 
         KafkaClusterOptions clusterOptions = this.GetAndValidateClusterOptions(cluster);
 
-        CachedSchemaRegistryClient? result;
-
-        lock (this._syncObj)
+        if (!this._registries.TryGetValue(clusterOptions, out CachedSchemaRegistryClient? result))
         {
-            if (!this._registries.TryGetValue(clusterOptions, out result))
-            {
-                result = new CachedSchemaRegistryClient(clusterOptions.SchemaRegistryConfig,
-                    clusterOptions.AuthenticationHeaderValueProvider);
+            ILogger logger = this._loggerFactory.CreateLogger(LoggerCategoryName);
 
-                this._registries.Add(clusterOptions, result);
+            lock (this._syncObj)
+            {
+                if (!this._registries.TryGetValue(clusterOptions, out result))
+                {
+                    try
+                    {
+                        result = new CachedSchemaRegistryClient(clusterOptions.SchemaRegistryConfig,
+                            clusterOptions.AuthenticationHeaderValueProvider);
+
+                        this._registries.Add(clusterOptions, result);
+
+                        logger.RegistryClientCreateOk(PrepareConfigForLogs(clusterOptions.SchemaRegistryConfig),
+                            clusterOptions.AuthenticationHeaderValueProvider?.GetType());
+                    }
+                    catch (Exception exception)
+                    {
+                        logger.RegistryClientCreateError(exception, clusterOptions.SchemaRegistryConfig,
+                            clusterOptions.AuthenticationHeaderValueProvider?.GetType());
+                        throw;
+                    }
+                }
             }
         }
 
@@ -265,7 +281,7 @@ private static void ValidateLogicalName(string? configName, string entityType)
         }
     }
 
-    private static IEnumerable<KeyValuePair<string, string>> PrepareConfigForLogs(Config config)
+    private static IEnumerable<KeyValuePair<string, string>> PrepareConfigForLogs(IEnumerable<KeyValuePair<string, string>> config)
     {
         return config.Select(x => Contains(x, "password") || Contains(x, "secret")
             ? new KeyValuePair<string, string>(x.Key, "*******")

src/Epam.Kafka/LogExtensions.cs

@@ -85,4 +85,19 @@ internal static partial void ProducerCreateOk(this ILogger logger, IEnumerable<K
         Message = "Producer create error. KeyType: {KeyType}, ValueType: {ValueType}, Config: {Config}.")]
     internal static partial void ProducerCreateError(this ILogger logger, Exception exception,
         IEnumerable<KeyValuePair<string, string>> config, Type keyType, Type valueType);
+
+    [LoggerMessage(
+        EventId = 203,
+        EventName = "RegistryClientCreated",
+        Level = LogLevel.Information,
+        Message = "Registry client created. HeaderProviderType: {HeaderProviderType}, Config: {Config}.")]
+    internal static partial void RegistryClientCreateOk(this ILogger logger, IEnumerable<KeyValuePair<string, string>> config, Type? headerProviderType);
+
+    [LoggerMessage(
+        EventId = 503,
+        EventName = "RegistryClientCreateError",
+        Level = LogLevel.Error,
+        Message = "Registry client create error. HeaderProviderType: {HeaderProviderType}, Config: {Config}.")]
+    internal static partial void RegistryClientCreateError(this ILogger logger, Exception exception,
+        IEnumerable<KeyValuePair<string, string>> config, Type? headerProviderType);
 }

src/Epam.Kafka/Options/Configuration/OptionsFromConfiguration.cs

@@ -1,9 +1,5 @@
 // Copyright © 2024 EPAM Systems
 
-#if !NET6_0_OR_GREATER
-using Epam.Kafka.Internals;
-#endif
-
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.Options;
 
@@ -35,7 +31,7 @@ public void Configure(string? name, TOptions options)
             return;
         }
 
-        IConfigurationSection section = this._configuration.GetSection(this.ParentSectionName).GetSection(name);
+        IConfigurationSection section = this._configuration.GetSection(this.ParentSectionName).GetSection(name!);
 
         if (section.Exists())
         {