Management of buffered data exchange based on serializer defined in configuration

src/documentation/articles/kafkadbcontext.md

@@ -8,8 +8,8 @@ _description: Describe what is and how use KafkaDbContext class from Entity Fram
 `KafkaDbContext` is a special class which helps to define the `DbContext` and use [Entity Framework Core](https://learn.microsoft.com/it-it/ef/core/) provider for [Apache Kafka](https://kafka.apache.org/):
 - `KafkaDbContext` inherits from `DbContext`: to define the model, and/or creating the database, see [getting started](https://docs.microsoft.com/ef/core/get-started/) in the docs and [KEFCore usage](usage.md)
 - `KafkaDbContext` defines the following properties:
-  - **KeySerializationType**: the .NET type to be used to allocate an external serializer for Apache Kafka record key
-  - **ValueSerializationType**: the .NET type to be used to allocate an external serializer for Apache Kafka record value
+  - **KeySerDesSelectorType**: the .NET type to be used to allocate an external serializer for Apache Kafka record key
+  - **ValueSerDesSelectorType**: the .NET type to be used to allocate an external serializer for Apache Kafka record value
   - **ValueContainerType**: the .NET type to be used to allocate an external container class for Apache Kafka record value
   - **UseNameMatching**: set to **false** to avoid Entity matching based on Name
   - **BootstrapServers**: the server hosting the broker of Apache Kafka
@@ -19,10 +19,11 @@ _description: Describe what is and how use KafkaDbContext class from Entity Fram
   - **DefaultReplicationFactor**: the replication factor to use when data are stored in Apache Kafka
   - **DefaultConsumerInstances**: the consumer instances to be allocated when UseCompactedReplicator is **true**
   - **UsePersistentStorage**: set to **true** to use a persistent storage between multiple application startup
+  - **UseEnumeratorWithPrefetch**: set to **true** to prefer enumerator instances able to do a prefetch on data speeding up execution, used if **UseKNetStreams** is **true** and **UseCompactedReplicator** is **false**
+  - **UseByteBufferDataTransfer**: set to **true** to prefer <see cref="Java.Nio.ByteBuffer"/> data exchange in serializer instances
   - **UseDeletePolicyForTopic**: set to **true** to enable [delete cleanup policy](https://kafka.apache.org/documentation/#topicconfigs_cleanup.policy)
   - **UseCompactedReplicator**: Use `KNetCompactedReplicator` instead of Apache Kafka Streams to manage data to or from topics
   - **UseKNetStreams**: Use KNet version of Apache Kafka Streams instead of standard Apache Kafka Streams, used if **UseCompactedReplicator** is **false**
-  - **UseEnumeratorWithPrefetch**: Setting this property to **true** the engine prefers to use enumerator instances able to do a prefetch on data speeding up execution, used if **UseKNetStreams** is **true** and **UseCompactedReplicator** is **false**
   - **ConsumerConfig**: parameters to use for Producer
   - **ProducerConfig**: parameters to use for Producer
   - **StreamsConfig**: parameters to use for Apche Kafka Streams application

src/documentation/articles/serialization.md

@@ -510,7 +510,7 @@ The extension converted this schema into code to speedup the exection of seriali
 ### How to use Protobuf 
 
 `KafkaDbContext` contains three properties can be used to override the default types:
-- **KeySerializationType**: set this value to `ProtobufKEFCoreSerDes..Key.BinaryRaw<>` or use `ProtobufKEFCoreSerDes.DefaultKeySerialization`, the type automatically manages simple or complex Primary Key
+- **KeySerializationType**: set this value to `ProtobufKEFCoreSerDes.Key.BinaryRaw<>` or use `ProtobufKEFCoreSerDes.DefaultKeySerialization`, the type automatically manages simple or complex Primary Key
 - **ValueSerializationType**: set this value to `ProtobufKEFCoreSerDes.ValueContainer.BinaryRaw<>` or use `ProtobufKEFCoreSerDes.DefaultValueContainerSerialization`
 - **ValueContainerType**: set this value to `ProtobufValueContainerRaw<>` or use `ProtobufKEFCoreSerDes.DefaultValueContainer`
 

src/net/KEFCore.SerDes.Avro.Compiler/KEFCore.SerDes.Avro.Compiler.csproj

@@ -23,7 +23,7 @@
 	</ItemGroup>
 
 	<ItemGroup>
-		<PackageReference Include="MASES.KNet.Serialization.Avro" Version="2.7.1" />
+		<PackageReference Include="MASES.KNet.Serialization.Avro" Version="2.7.2" />
 	</ItemGroup>
 
 	<ItemGroup>

src/net/KEFCore.SerDes.Avro/AvroValueContainer.cs

@@ -21,6 +21,7 @@
 #nullable enable
 
 using Avro;
+using MASES.KNet.Serialization;
 
 namespace MASES.EntityFrameworkCore.KNet.Serialization.Avro.Storage;
 
@@ -44,7 +45,7 @@ public partial class AvroValueContainer<TKey> : AvroValueContainer, IValueContai
     public AvroValueContainer(IEntityType tName, object[] rData)
     {
         EntityName = tName.Name;
-        ClrType = tName.ClrType.FullName!;
+        ClrType = tName.ClrType?.ToAssemblyQualified()!;
         Data = new List<PropertyDataRecord>();
         foreach (var item in tName.GetProperties())
         {
@@ -53,7 +54,7 @@ public AvroValueContainer(IEntityType tName, object[] rData)
             {
                 PropertyIndex = index,
                 PropertyName = item.Name,
-                ClrType = item.ClrType?.FullName,
+                ClrType = item.ClrType?.ToAssemblyQualified(),
                 Value = rData[index]
             };
             Data.Add(pRecord);

src/net/KEFCore.SerDes.Protobuf/Generated/GenericValue.cs

@@ -772,7 +772,11 @@ public void MergeFrom(pb::CodedInputStream input) {
     #else
       uint tag;
       while ((tag = input.ReadTag()) != 0) {
-        switch(tag) {
+      if ((tag & 7) == 4) {
+        // Abort on any end group tag.
+        return;
+      }
+      switch(tag) {
           default:
             _unknownFields = pb::UnknownFieldSet.MergeFieldFrom(_unknownFields, input);
             break;
@@ -846,7 +850,11 @@ public void MergeFrom(pb::CodedInputStream input) {
     void pb::IBufferMessage.InternalMergeFrom(ref pb::ParseContext input) {
       uint tag;
       while ((tag = input.ReadTag()) != 0) {
-        switch(tag) {
+      if ((tag & 7) == 4) {
+        // Abort on any end group tag.
+        return;
+      }
+      switch(tag) {
           default:
             _unknownFields = pb::UnknownFieldSet.MergeFieldFrom(_unknownFields, ref input);
             break;

src/net/KEFCore.SerDes.Protobuf/Generated/KeyContainer.cs

@@ -190,7 +190,11 @@ public void MergeFrom(pb::CodedInputStream input) {
     #else
       uint tag;
       while ((tag = input.ReadTag()) != 0) {
-        switch(tag) {
+      if ((tag & 7) == 4) {
+        // Abort on any end group tag.
+        return;
+      }
+      switch(tag) {
           default:
             _unknownFields = pb::UnknownFieldSet.MergeFieldFrom(_unknownFields, input);
             break;
@@ -209,7 +213,11 @@ public void MergeFrom(pb::CodedInputStream input) {
     void pb::IBufferMessage.InternalMergeFrom(ref pb::ParseContext input) {
       uint tag;
       while ((tag = input.ReadTag()) != 0) {
-        switch(tag) {
+      if ((tag & 7) == 4) {
+        // Abort on any end group tag.
+        return;
+      }
+      switch(tag) {
           default:
             _unknownFields = pb::UnknownFieldSet.MergeFieldFrom(_unknownFields, ref input);
             break;
@@ -386,7 +394,11 @@ public void MergeFrom(pb::CodedInputStream input) {
     #else
       uint tag;
       while ((tag = input.ReadTag()) != 0) {
-        switch(tag) {
+      if ((tag & 7) == 4) {
+        // Abort on any end group tag.
+        return;
+      }
+      switch(tag) {
           default:
             _unknownFields = pb::UnknownFieldSet.MergeFieldFrom(_unknownFields, input);
             break;
@@ -408,7 +420,11 @@ public void MergeFrom(pb::CodedInputStream input) {
     void pb::IBufferMessage.InternalMergeFrom(ref pb::ParseContext input) {
       uint tag;
       while ((tag = input.ReadTag()) != 0) {
-        switch(tag) {
+      if ((tag & 7) == 4) {
+        // Abort on any end group tag.
+        return;
+      }
+      switch(tag) {
           default:
             _unknownFields = pb::UnknownFieldSet.MergeFieldFrom(_unknownFields, ref input);
             break;

src/net/KEFCore.SerDes.Protobuf/Generated/ValueContainer.cs

@@ -294,7 +294,11 @@ public void MergeFrom(pb::CodedInputStream input) {
     #else
       uint tag;
       while ((tag = input.ReadTag()) != 0) {
-        switch(tag) {
+      if ((tag & 7) == 4) {
+        // Abort on any end group tag.
+        return;
+      }
+      switch(tag) {
           default:
             _unknownFields = pb::UnknownFieldSet.MergeFieldFrom(_unknownFields, input);
             break;
@@ -328,7 +332,11 @@ public void MergeFrom(pb::CodedInputStream input) {
     void pb::IBufferMessage.InternalMergeFrom(ref pb::ParseContext input) {
       uint tag;
       while ((tag = input.ReadTag()) != 0) {
-        switch(tag) {
+      if ((tag & 7) == 4) {
+        // Abort on any end group tag.
+        return;
+      }
+      switch(tag) {
           default:
             _unknownFields = pb::UnknownFieldSet.MergeFieldFrom(_unknownFields, ref input);
             break;
@@ -564,7 +572,11 @@ public void MergeFrom(pb::CodedInputStream input) {
     #else
       uint tag;
       while ((tag = input.ReadTag()) != 0) {
-        switch(tag) {
+      if ((tag & 7) == 4) {
+        // Abort on any end group tag.
+        return;
+      }
+      switch(tag) {
           default:
             _unknownFields = pb::UnknownFieldSet.MergeFieldFrom(_unknownFields, input);
             break;
@@ -591,7 +603,11 @@ public void MergeFrom(pb::CodedInputStream input) {
     void pb::IBufferMessage.InternalMergeFrom(ref pb::ParseContext input) {
       uint tag;
       while ((tag = input.ReadTag()) != 0) {
-        switch(tag) {
+      if ((tag & 7) == 4) {
+        // Abort on any end group tag.
+        return;
+      }
+      switch(tag) {
           default:
             _unknownFields = pb::UnknownFieldSet.MergeFieldFrom(_unknownFields, ref input);
             break;

src/net/KEFCore.SerDes.Protobuf/KEFCore.SerDes.Protobuf.csproj

@@ -42,8 +42,8 @@
 
 	<ItemGroup>
 		<ProjectReference Include="..\KEFCore.SerDes\KEFCore.SerDes.csproj" />
-		<PackageReference Include="Google.Protobuf" Version="3.26.1" />
-		<PackageReference Include="Google.Protobuf.Tools" Version="3.26.1">
+		<PackageReference Include="Google.Protobuf" Version="3.27.0" />
+		<PackageReference Include="Google.Protobuf.Tools" Version="3.27.0">
 			<PrivateAssets>all</PrivateAssets>
 			<IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
 		</PackageReference>