feat: upgrade to Confluent.Kafka.* 2.8.0

MiguelCosta · brmagadutra · commit d01ca999fd65 · 2025-02-06T15:57:05.000Z
diff --git a/src/KafkaFlow.SchemaRegistry/KafkaFlow.SchemaRegistry.csproj b/src/KafkaFlow.SchemaRegistry/KafkaFlow.SchemaRegistry.csproj
@@ -8,7 +8,7 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="Confluent.SchemaRegistry" Version="2.4.0" />
+    <PackageReference Include="Confluent.SchemaRegistry" Version="2.8.0" />
   </ItemGroup>
 
   <ItemGroup>
diff --git a/src/KafkaFlow.Serializer.JsonCore/JsonCoreSerializer.cs b/src/KafkaFlow.Serializer.JsonCore/JsonCoreSerializer.cs
@@ -1,4 +1,5 @@
-﻿using System.IO;
+﻿using System;
+using System.IO;
 using System.Text.Json;
 using System.Threading.Tasks;
 
@@ -52,6 +53,11 @@ public JsonCoreSerializer()
     /// <inheritdoc/>
     public Task SerializeAsync(object message, Stream output, ISerializerContext context)
     {
+        if (message is null || message == Array.Empty<byte>())
+        {
+            return Task.CompletedTask;
+        }
+
         using var writer = new Utf8JsonWriter(output, _writerOptions);
 
         JsonSerializer.Serialize(writer, message, _serializerOptions);
diff --git a/src/KafkaFlow.Serializer.ProtobufNet/KafkaFlow.Serializer.ProtobufNet.csproj b/src/KafkaFlow.Serializer.ProtobufNet/KafkaFlow.Serializer.ProtobufNet.csproj
@@ -8,7 +8,7 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="protobuf-net" Version="3.2.45" />
+    <PackageReference Include="protobuf-net" Version="3.2.12" />
   </ItemGroup>
 
   <ItemGroup>
diff --git a/src/KafkaFlow.Serializer.SchemaRegistry.ConfluentAvro/ConfluentAvroSerializer.cs b/src/KafkaFlow.Serializer.SchemaRegistry.ConfluentAvro/ConfluentAvroSerializer.cs
@@ -40,7 +40,8 @@ public Task SerializeAsync(object message, Stream output, ISerializerContext con
                 () => Activator.CreateInstance(
                     typeof(AvroSerializer<>).MakeGenericType(message.GetType()),
                     _schemaRegistryClient,
-                    _serializerConfig))
+                    _serializerConfig,
+                    null))
             .SerializeAsync(message, output, context);
     }
 }
diff --git a/src/KafkaFlow.Serializer.SchemaRegistry.ConfluentAvro/KafkaFlow.Serializer.SchemaRegistry.ConfluentAvro.csproj b/src/KafkaFlow.Serializer.SchemaRegistry.ConfluentAvro/KafkaFlow.Serializer.SchemaRegistry.ConfluentAvro.csproj
@@ -14,9 +14,9 @@
 
   <ItemGroup>
     <PackageReference Include="Apache.Avro" Version="1.12.0" />
-    <PackageReference Include="Confluent.Kafka" Version="2.4.0" />
-    <PackageReference Include="Confluent.SchemaRegistry" Version="2.4.0" />
-    <PackageReference Include="Confluent.SchemaRegistry.Serdes.Avro" Version="2.4.0" />
+    <PackageReference Include="Confluent.Kafka" Version="2.8.0" />
+    <PackageReference Include="Confluent.SchemaRegistry" Version="2.8.0" />
+    <PackageReference Include="Confluent.SchemaRegistry.Serdes.Avro" Version="2.8.0" />
   </ItemGroup>
 
 </Project>
diff --git a/src/KafkaFlow.Serializer.SchemaRegistry.ConfluentJson/ConfluentJsonSerializer.cs b/src/KafkaFlow.Serializer.SchemaRegistry.ConfluentJson/ConfluentJsonSerializer.cs
@@ -59,7 +59,8 @@ public Task SerializeAsync(object message, Stream output, ISerializerContext con
                     typeof(JsonSerializer<>).MakeGenericType(message.GetType()),
                     _schemaRegistryClient,
                     _serializerConfig,
-                    _schemaGeneratorSettings))
+                    _schemaGeneratorSettings,
+                    null))
             .SerializeAsync(message, output, context);
     }
 }
diff --git a/src/KafkaFlow.Serializer.SchemaRegistry.ConfluentJson/KafkaFlow.Serializer.SchemaRegistry.ConfluentJson.csproj b/src/KafkaFlow.Serializer.SchemaRegistry.ConfluentJson/KafkaFlow.Serializer.SchemaRegistry.ConfluentJson.csproj
@@ -8,7 +8,7 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="Confluent.SchemaRegistry.Serdes.Json" Version="2.4.0" />
+    <PackageReference Include="Confluent.SchemaRegistry.Serdes.Json" Version="2.8.0" />
   </ItemGroup>
 
   <ItemGroup>
diff --git a/src/KafkaFlow.Serializer.SchemaRegistry.ConfluentProtobuf/ConfluentProtobufSerializer.cs b/src/KafkaFlow.Serializer.SchemaRegistry.ConfluentProtobuf/ConfluentProtobufSerializer.cs
@@ -38,7 +38,8 @@ public Task SerializeAsync(object message, Stream output, ISerializerContext con
                 () => Activator.CreateInstance(
                     typeof(ProtobufSerializer<>).MakeGenericType(message.GetType()),
                     _schemaRegistryClient,
-                    _serializerConfig))
+                    _serializerConfig,
+                    null))
             .SerializeAsync(message, output, context);
     }
 }
diff --git a/src/KafkaFlow.Serializer.SchemaRegistry.ConfluentProtobuf/ConfluentProtobufTypeNameResolver.cs b/src/KafkaFlow.Serializer.SchemaRegistry.ConfluentProtobuf/ConfluentProtobufTypeNameResolver.cs
@@ -1,8 +1,8 @@
+extern alias GoogleProto;
+
 using System.Linq;
 using System.Threading.Tasks;
 using Confluent.SchemaRegistry;
-using Google.Protobuf;
-using Google.Protobuf.Reflection;
 
 namespace KafkaFlow;
 
@@ -19,7 +19,8 @@ public async Task<string> ResolveAsync(int id)
     {
         var schemaString = (await _client.GetSchemaAsync(id, "serialized")).SchemaString;
 
-        var protoFields = FileDescriptorProto.Parser.ParseFrom(ByteString.FromBase64(schemaString));
+        // Issue: https://github.com/confluentinc/confluent-kafka-dotnet/issues/2409
+        var protoFields = GoogleProto::Google.Protobuf.Reflection.FileDescriptorProto.Parser.ParseFrom(GoogleProto::Google.Protobuf.ByteString.FromBase64(schemaString));
 
         return $"{protoFields.Package}.{protoFields.MessageType.FirstOrDefault()?.Name}";
     }
diff --git a/src/KafkaFlow.Serializer.SchemaRegistry.ConfluentProtobuf/KafkaFlow.Serializer.SchemaRegistry.ConfluentProtobuf.csproj b/src/KafkaFlow.Serializer.SchemaRegistry.ConfluentProtobuf/KafkaFlow.Serializer.SchemaRegistry.ConfluentProtobuf.csproj
@@ -8,7 +8,13 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="Confluent.SchemaRegistry.Serdes.Protobuf" Version="2.4.0" />
+    <PackageReference Include="Confluent.SchemaRegistry.Serdes.Protobuf" Version="2.8.0" />
+    <PackageReference Include="Google.Protobuf" Version="3.26.1">
+      <Aliases>GoogleProto</Aliases>
+    </PackageReference>
+    <PackageReference Include="protobuf-net.Reflection" Version="3.2.12">
+      <Aliases>PbNet</Aliases>
+    </PackageReference>
   </ItemGroup>
 
   <ItemGroup>
diff --git a/src/KafkaFlow/KafkaFlow.csproj b/src/KafkaFlow/KafkaFlow.csproj
@@ -7,7 +7,7 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="Confluent.Kafka" Version="2.4.0" />
+    <PackageReference Include="Confluent.Kafka" Version="2.8.0" />
     <PackageReference Include="Microsoft.Bcl.AsyncInterfaces" Version="8.0.0" />
     <PackageReference Include="Microsoft.IO.RecyclableMemoryStream" Version="3.0.1" />
     <PackageReference Include="System.Threading.Channels" Version="8.0.0" />

Original file line number	Diff line number	Diff line change
`@@ -40,7 +40,8 @@ public Task SerializeAsync(object message, Stream output, ISerializerContext con`
`40`	`40`	`() => Activator.CreateInstance(`
`41`	`41`	`typeof(AvroSerializer<>).MakeGenericType(message.GetType()),`
`42`	`42`	`_schemaRegistryClient,`
`43`		`- _serializerConfig))`
	`43`	`+ _serializerConfig,`
	`44`	`+ null))`
`44`	`45`	`.SerializeAsync(message, output, context);`
`45`	`46`	`}`
`46`	`47`	`}`
Original file line number	Diff line number	Diff line change
`@@ -59,7 +59,8 @@ public Task SerializeAsync(object message, Stream output, ISerializerContext con`
`59`	`59`	`typeof(JsonSerializer<>).MakeGenericType(message.GetType()),`
`60`	`60`	`_schemaRegistryClient,`
`61`	`61`	`_serializerConfig,`
`62`		`- _schemaGeneratorSettings))`
	`62`	`+ _schemaGeneratorSettings,`
	`63`	`+ null))`
`63`	`64`	`.SerializeAsync(message, output, context);`
`64`	`65`	`}`
`65`	`66`	`}`
Original file line number	Diff line number	Diff line change
`@@ -38,7 +38,8 @@ public Task SerializeAsync(object message, Stream output, ISerializerContext con`
`38`	`38`	`() => Activator.CreateInstance(`
`39`	`39`	`typeof(ProtobufSerializer<>).MakeGenericType(message.GetType()),`
`40`	`40`	`_schemaRegistryClient,`
`41`		`- _serializerConfig))`
	`41`	`+ _serializerConfig,`
	`42`	`+ null))`
`42`	`43`	`.SerializeAsync(message, output, context);`
`43`	`44`	`}`
`44`	`45`	`}`