add null checks for public methods + ConfigureAwait

.github/linters/.jscpd.json

@@ -0,0 +1,7 @@
+{
+  "threshold": 16,
+  "ignore": [
+    "**/.github/**",
+    "**/tests/**"
+  ]
+}

src/Confluent.Kafka.Extensions.Diagnostics/ActivityDiagnosticsHelper.cs

@@ -53,37 +53,45 @@ internal static class ActivityDiagnosticsHelper
     internal static Activity? StartConsumeActivity<TKey, TValue>(TopicPartition partition,
         Message<TKey, TValue> message)
     {
-        var activity = ActivitySource.CreateActivity("Confluent.Kafka.Consume", ActivityKind.Consumer,
-            default(ActivityContext), ActivityTags(partition)!);
-
-        if (activity != null)
+        try
         {
-            var traceParentHeader = message.Headers?.FirstOrDefault(x => x.Key == TraceParentHeaderName);
-            var traceStateHeader = message.Headers?.FirstOrDefault(x => x.Key == TraceStateHeaderName);
-
-            var traceParent = traceParentHeader != null
-                ? Encoding.UTF8.GetString(traceParentHeader.GetValueBytes())
-                : null;
-            var traceState = traceStateHeader != null
-                ? Encoding.UTF8.GetString(traceStateHeader.GetValueBytes())
-                : null;
+            var activity = ActivitySource.CreateActivity("Confluent.Kafka.Consume", ActivityKind.Consumer,
+                default(ActivityContext), ActivityTags(partition)!);
 
-            if (ActivityContext.TryParse(traceParent, traceState, out var activityContext))
+            if (activity != null)
             {
-                activity.SetParentId(activityContext.TraceId, activityContext.SpanId, activityContext.TraceFlags);
-                activity.TraceStateString = activityContext.TraceState;
+                var traceParentHeader = message.Headers?.FirstOrDefault(x => x.Key == TraceParentHeaderName);
+                var traceStateHeader = message.Headers?.FirstOrDefault(x => x.Key == TraceStateHeaderName);
+
+                var traceParent = traceParentHeader != null
+                    ? Encoding.UTF8.GetString(traceParentHeader.GetValueBytes())
+                    : null;
+                var traceState = traceStateHeader != null
+                    ? Encoding.UTF8.GetString(traceStateHeader.GetValueBytes())
+                    : null;
+
+                if (ActivityContext.TryParse(traceParent, traceState, out var activityContext))
+                {
+                    activity.SetParentId(activityContext.TraceId, activityContext.SpanId, activityContext.TraceFlags);
+                    activity.TraceStateString = activityContext.TraceState;
+                }
+
+                if (activity.IsAllDataRequested)
+                {
+                    SetActivityTags(activity, message);
+                }
+
+                activity.Start();
             }
 
-            if (activity.IsAllDataRequested)
-            {
-                SetActivityTags(activity, message);
-            }
 
-            activity.Start();
+            return activity;
+        }
+        catch
+        {
+            // ignore
+            return null;
         }
-
-
-        return activity;
     }
 
     private static void SetActivityTags<TKey, TValue>(Activity activity, Message<TKey, TValue> message)

src/Confluent.Kafka.Extensions.Diagnostics/ConsumerExtensions.cs

@@ -11,13 +11,16 @@ public static class ConsumerExtensions
     public static async Task ConsumeWithInstrumentation<TKey, TValue>(this IConsumer<TKey, TValue> consumer,
         Func<ConsumeResult<TKey, TValue>?, CancellationToken, Task> action, CancellationToken cancellationToken)
     {
+        if (consumer == null) throw new ArgumentNullException(nameof(consumer));
+        if (action == null) throw new ArgumentNullException(nameof(action));
+
         var result = consumer.Consume(cancellationToken);
 
         var activity = ActivityDiagnosticsHelper.StartConsumeActivity(result.TopicPartition, result.Message);
 
         try
         {
-            await action(result, cancellationToken);
+            await action(result, cancellationToken).ConfigureAwait(false);
         }
         finally
         {
@@ -30,15 +33,19 @@ public static async Task ConsumeWithInstrumentation<TKey, TValue>(this IConsumer
     /// </summary>
     public static async Task<TResult> ConsumeWithInstrumentation<TKey, TValue, TResult>(
         this IConsumer<TKey, TValue> consumer,
-        Func<ConsumeResult<TKey, TValue>?, CancellationToken, Task<TResult>> action, CancellationToken cancellationToken)
+        Func<ConsumeResult<TKey, TValue>?, CancellationToken, Task<TResult>> action,
+        CancellationToken cancellationToken)
     {
+        if (consumer == null) throw new ArgumentNullException(nameof(consumer));
+        if (action == null) throw new ArgumentNullException(nameof(action));
+
         var result = consumer.Consume(cancellationToken);
 
         var activity = ActivityDiagnosticsHelper.StartConsumeActivity(result.TopicPartition, result.Message);
 
         try
         {
-            return await action(result, cancellationToken);
+            return await action(result, cancellationToken).ConfigureAwait(false);
         }
         finally
         {
@@ -52,6 +59,9 @@ public static async Task<TResult> ConsumeWithInstrumentation<TKey, TValue, TResu
     public static void ConsumeWithInstrumentation<TKey, TValue>(this IConsumer<TKey, TValue> consumer,
         Action<ConsumeResult<TKey, TValue>?> action, int millisecondsTimeout)
     {
+        if (consumer == null) throw new ArgumentNullException(nameof(consumer));
+        if (action == null) throw new ArgumentNullException(nameof(action));
+
         var result = consumer.Consume(millisecondsTimeout);
 
         var activity = ActivityDiagnosticsHelper.StartConsumeActivity(result.TopicPartition, result.Message);

src/Confluent.Kafka.Extensions.Diagnostics/InstrumentedProducer.cs

@@ -30,7 +30,8 @@ public async Task<DeliveryResult<TKey, TValue>> ProduceAsync(
         try
         {
             // todo: get delivery result and put it into the activity
-            return await _producerImplementation.ProduceAsync(topicPartition, message, cancellationToken);
+            return await _producerImplementation.ProduceAsync(topicPartition, message, cancellationToken)
+                .ConfigureAwait(false);
         }
         finally
         {

src/Confluent.Kafka.Extensions.Diagnostics/ProducerBuilderExtensions.cs

@@ -11,6 +11,8 @@ public static class ProducerBuilderExtensions
     public static IProducer<TKey, TValue> BuildWithInstrumentation<TKey, TValue>(
         this ProducerBuilder<TKey, TValue> producerBuilder)
     {
+        if (producerBuilder == null) throw new ArgumentNullException(nameof(producerBuilder));
+
         return new InstrumentedProducer<TKey, TValue>(producerBuilder.Build());
     }
 }