Merge branch '1.0.x'

Author: mhowlett

CHANGELOG.md

@@ -3,7 +3,7 @@
 ## New Features
 
 - Added GET subject versions to the cached schema registry client.
-- References librdkafka.redist [1.0.0-RC8](https://github.com/edenhill/librdkafka/releases/tag/v1.0.0-RC8)
+- References librdkafka.redist [1.0.0-RC9](https://github.com/edenhill/librdkafka/releases/tag/v1.0.0-RC9)
   - supports apline linux out-of-the-box.
   - fallback support (that excludes security features) for most linux distributions previously unsuppored out-of-the-box.
   - fixed a dependency issue on MacOS

src/Confluent.Kafka/Confluent.Kafka.csproj

@@ -19,7 +19,7 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="librdkafka.redist" Version="1.0.0-RC8">
+    <PackageReference Include="librdkafka.redist" Version="1.0.0-RC9">
         <PrivateAssets Condition="'$(TargetFrameworkIdentifier)' == '.NETFramework'">None</PrivateAssets>
     </PackageReference>
     <PackageReference Include="System.Memory" Version="4.5.0" />

src/Confluent.Kafka/Consumer.cs

@@ -422,7 +422,7 @@ public void StoreOffset(ConsumeResult<TKey, TValue> result)
         ///     `enable.auto.offset.store` must be set to "false" when using this API.
         /// </remarks>
         /// <param name="offset">
-        ///     The offset to be commited.
+        ///     The offset to be committed.
         /// </param>
         /// <exception cref="Confluent.Kafka.KafkaException">
         ///     Thrown if the request failed.
@@ -1014,7 +1014,7 @@ private ConsumeResult<K, V> ConsumeImpl<K,V>(
                             new SerializationContext(MessageComponentType.Key, topic));
                     }
                 }
-                catch (Exception exception)
+                catch (Exception ex)
                 {
                     throw new ConsumeException(
                         new ConsumeResult<byte[], byte[]>
@@ -1030,7 +1030,7 @@ private ConsumeResult<K, V> ConsumeImpl<K,V>(
                             IsPartitionEOF = false
                         },
                         new Error(ErrorCode.Local_KeyDeserialization),
-                        exception);
+                        ex);
                 }
 
                 V val;
@@ -1046,7 +1046,7 @@ private ConsumeResult<K, V> ConsumeImpl<K,V>(
                             new SerializationContext(MessageComponentType.Value, topic));
                     }
                 }
-                catch (Exception exception)
+                catch (Exception ex)
                 {
                     throw new ConsumeException(
                         new ConsumeResult<byte[], byte[]>
@@ -1062,7 +1062,7 @@ private ConsumeResult<K, V> ConsumeImpl<K,V>(
                             IsPartitionEOF = false
                         },
                         new Error(ErrorCode.Local_ValueDeserialization),
-                        exception);
+                        ex);
                 }
 
                 return new ConsumeResult<K, V> 
@@ -1100,19 +1100,35 @@ private ConsumeResult<TKey, TValue> ConsumeViaBytes(int millisecondsTimeout)
                 };
             }
 
-            TKey key = keyDeserializer != null
-                ? keyDeserializer.Deserialize(rawResult.Key, rawResult.Key == null, new SerializationContext(MessageComponentType.Key, rawResult.Topic))
-                : asyncKeyDeserializer.DeserializeAsync(new ReadOnlyMemory<byte>(rawResult.Key), rawResult.Key == null, new SerializationContext(MessageComponentType.Key, rawResult.Topic))
-                    .ConfigureAwait(continueOnCapturedContext: false)
-                    .GetAwaiter()
-                    .GetResult();
+            TKey key;
+            try
+            {
+                key = keyDeserializer != null
+                    ? keyDeserializer.Deserialize(rawResult.Key, rawResult.Key == null, new SerializationContext(MessageComponentType.Key, rawResult.Topic))
+                    : Task.Run(async () => await asyncKeyDeserializer.DeserializeAsync(new ReadOnlyMemory<byte>(rawResult.Key), rawResult.Key == null, new SerializationContext(MessageComponentType.Key, rawResult.Topic)))
+                        .ConfigureAwait(continueOnCapturedContext: false)
+                        .GetAwaiter()
+                        .GetResult();
+            }
+            catch (Exception ex)
+            {
+                throw new ConsumeException(rawResult, new Error(ErrorCode.Local_KeyDeserialization), ex);
+            }
 
-            TValue val = valueDeserializer != null
-                ? valueDeserializer.Deserialize(rawResult.Value, rawResult.Value == null, new SerializationContext(MessageComponentType.Value, rawResult.Topic))
-                : asyncValueDeserializer.DeserializeAsync(new ReadOnlyMemory<byte>(rawResult.Value), rawResult == null, new SerializationContext(MessageComponentType.Value, rawResult.Topic))
-                    .ConfigureAwait(continueOnCapturedContext: false)
-                    .GetAwaiter()
-                    .GetResult();
+            TValue val;
+            try
+            {
+                val = valueDeserializer != null
+                    ? valueDeserializer.Deserialize(rawResult.Value, rawResult.Value == null, new SerializationContext(MessageComponentType.Value, rawResult.Topic))
+                    : Task.Run(async () => await asyncValueDeserializer.DeserializeAsync(new ReadOnlyMemory<byte>(rawResult.Value), rawResult == null, new SerializationContext(MessageComponentType.Value, rawResult.Topic)))
+                        .ConfigureAwait(continueOnCapturedContext: false)
+                        .GetAwaiter()
+                        .GetResult();
+            }
+            catch (Exception ex)
+            {
+                throw new ConsumeException(rawResult, new Error(ErrorCode.Local_ValueDeserialization), ex);
+            }
 
             return new ConsumeResult<TKey, TValue>
             {
@@ -1140,10 +1156,18 @@ private ConsumeResult<TKey, TValue> ConsumeViaBytes(int millisecondsTimeout)
         ///     The consume result.
         /// </returns>
         /// <remarks>
-        ///     OnPartitionsAssigned/Revoked and OnOffsetsCommitted events may
+        ///     The partitions assigned/revoked and offsets committed handlers may
         ///     be invoked as a side-effect of calling this method (on the same
         ///     thread).
         /// </remarks>
+        /// <exception cref="ConsumeException">
+        ///     Thrown when a call to this method is unsuccessful for any reason
+        ///     (except cancellation by user). Inspect the Error property of the
+        ///     exception for detailed information.
+        /// </exception>
+        /// <exception cref="OperationCanceledException">
+        ///     Thrown on cancellation.
+        /// </exception>
         public ConsumeResult<TKey, TValue> Consume(CancellationToken cancellationToken = default(CancellationToken))
         {
             while (true)
@@ -1172,10 +1196,14 @@ private ConsumeResult<TKey, TValue> ConsumeViaBytes(int millisecondsTimeout)
         ///     The consume result.
         /// </returns>
         /// <remarks>
-        ///     OnPartitionsAssigned/Revoked and OnOffsetsCommitted events may
+        ///     The partitions assigned/revoked and offsets committed handlers may
         ///     be invoked as a side-effect of calling this method (on the same
         ///     thread).
         /// </remarks>
+        /// <exception cref="ConsumeException">
+        ///     Thrown when a call to this method is unsuccessful for any reason.
+        ///     Inspect the Error property of the exception for detailed information.
+        /// </exception>
         public ConsumeResult<TKey, TValue> Consume(TimeSpan timeout)
             => (keyDeserializer != null && valueDeserializer != null)
                 ? ConsumeImpl<TKey, TValue>(timeout.TotalMillisecondsAsInt(), keyDeserializer, valueDeserializer) // fast path for simple case

src/Confluent.Kafka/DeserializationException.cs

@@ -39,28 +39,17 @@ public string Deserialize(ReadOnlySpan<byte> data, bool isNull, SerializationCon
                     return null;
                 }
 
-                try
-                {
-                    #if NETCOREAPP2_1
-                        return Encoding.UTF8.GetString(data);
-                    #else
-                        return Encoding.UTF8.GetString(data.ToArray());
-                    #endif
-                }
-                catch (Exception e)
-                {
-                    throw new DeserializationException("Error occured deserializing UTF8 string value", e);
-                }
+                #if NETCOREAPP2_1
+                    return Encoding.UTF8.GetString(data);
+                #else
+                    return Encoding.UTF8.GetString(data.ToArray());
+                #endif
             }
         }
 
         /// <summary>
         ///     Null value deserializer.
         /// </summary>
-        /// <remarks>
-        ///     Unexpected input will result in a
-        ///     <see cref="DeserializationException" />.
-        /// </remarks>
         public static IDeserializer<Null> Null = new NullDeserializer();
 
         private class NullDeserializer : IDeserializer<Null>
@@ -69,7 +58,7 @@ public Null Deserialize(ReadOnlySpan<byte> data, bool isNull, SerializationConte
             {
                 if (!isNull)
                 {
-                    throw new DeserializationException("Deserializer<Null> may only be used to deserialize data that is null.");
+                    throw new ArgumentException("Deserializer<Null> may only be used to deserialize data that is null.");
                 }
 
                 return null;
@@ -98,12 +87,12 @@ public long Deserialize(ReadOnlySpan<byte> data, bool isNull, SerializationConte
             {
                 if (isNull)
                 {
-                    throw new DeserializationException($"Null data encountered deserializing Int64 value.");
+                    throw new ArgumentNullException($"Null data encountered deserializing Int64 value.");
                 }
 
                 if (data.Length != 8)
                 {
-                    throw new DeserializationException($"Deserializer<Long> encountered data of length {data.Length}. Expecting data length to be 8.");
+                    throw new ArgumentException($"Deserializer<Long> encountered data of length {data.Length}. Expecting data length to be 8.");
                 }
 
                 // network byte order -> big endian -> most significant byte in the smallest address.
@@ -130,12 +119,12 @@ public int Deserialize(ReadOnlySpan<byte> data, bool isNull, SerializationContex
             {
                 if (isNull)
                 {
-                    throw new DeserializationException($"Null data encountered deserializing an Int32 value");
+                    throw new ArgumentNullException($"Null data encountered deserializing Int32 value");
                 }
 
                 if (data.Length != 4)
                 {
-                    throw new DeserializationException($"Deserializer<Int32> encountered data of length {data.Length}. Expecting data length to be 4.");
+                    throw new ArgumentException($"Deserializer<Int32> encountered data of length {data.Length}. Expecting data length to be 4.");
                 }
 
                 // network byte order -> big endian -> most significant byte in the smallest address.
@@ -158,12 +147,12 @@ public float Deserialize(ReadOnlySpan<byte> data, bool isNull, SerializationCont
             {
                 if (isNull)
                 {
-                    throw new DeserializationException($"Null data encountered deserializing an float value.");
+                    throw new ArgumentNullException($"Null data encountered deserializing float value.");
                 }
 
                 if (data.Length != 4)
                 {
-                    throw new DeserializationException($"Deserializer<float> encountered data of length {data.Length}. Expecting data length to be 4.");
+                    throw new ArgumentException($"Deserializer<float> encountered data of length {data.Length}. Expecting data length to be 4.");
                 }
 
                 // network byte order -> big endian -> most significant byte in the smallest address.
@@ -182,18 +171,11 @@ public float Deserialize(ReadOnlySpan<byte> data, bool isNull, SerializationCont
                 }
                 else
                 {
-                    try
-                    {
-                        #if NETCOREAPP2_1
-                            return BitConverter.ToSingle(data);
-                        #else
-                            return BitConverter.ToSingle(data.ToArray(), 0);
-                        #endif
-                    }
-                    catch (Exception e)
-                    {
-                        throw new DeserializationException("Error occured deserializing float value.", e);
-                    }
+                    #if NETCOREAPP2_1
+                        return BitConverter.ToSingle(data);
+                    #else
+                        return BitConverter.ToSingle(data.ToArray(), 0);
+                    #endif
                 }
             }
         }
@@ -209,12 +191,12 @@ public double Deserialize(ReadOnlySpan<byte> data, bool isNull, SerializationCon
             {
                 if (isNull)
                 {
-                    throw new DeserializationException($"Null data encountered deserializing an double value.");
+                    throw new ArgumentNullException($"Null data encountered deserializing double value.");
                 }
 
                 if (data.Length != 8)
                 {
-                    throw new DeserializationException($"Deserializer<double> encountered data of length {data.Length}. Expecting data length to be 8.");
+                    throw new ArgumentException($"Deserializer<double> encountered data of length {data.Length}. Expecting data length to be 8.");
                 }
 
                 // network byte order -> big endian -> most significant byte in the smallest address.
@@ -237,18 +219,11 @@ public double Deserialize(ReadOnlySpan<byte> data, bool isNull, SerializationCon
                 }
                 else
                 {
-                    try
-                    {
-                        #if NETCOREAPP2_1
-                                        return BitConverter.ToDouble(data);
-                        #else
-                                        return BitConverter.ToDouble(data.ToArray(), 0);
-                        #endif
-                    }
-                    catch (Exception e)
-                    {
-                        throw new DeserializationException("Error occured deserializing double value.", e);
-                    }
+                    #if NETCOREAPP2_1
+                                    return BitConverter.ToDouble(data);
+                    #else
+                                    return BitConverter.ToDouble(data.ToArray(), 0);
+                    #endif
                 }
             }
         }