librdkafka v2.1.0 (#2035)

emasab · web-flow · commit de70d7d11011 · 2023-04-06T19:14:09.000+05:30
* librdkafka v2.1.0

* Fix KIP-320 add log truncation error code

* Fix failing Consumer_Poll_MessageError test
diff --git a/src/Confluent.Kafka/Config_gen.cs b/src/Confluent.Kafka/Config_gen.cs
@@ -1,4 +1,4 @@
-// *** Auto-generated from librdkafka v2.1.0-RC3 *** - do not modify manually.
+// *** Auto-generated from librdkafka v2.1.0 *** - do not modify manually.
 //
 // Copyright 2018-2022 Confluent Inc.
 //
diff --git a/src/Confluent.Kafka/Confluent.Kafka.csproj b/src/Confluent.Kafka/Confluent.Kafka.csproj
@@ -12,7 +12,7 @@
     <PackageId>Confluent.Kafka</PackageId>
     <Title>Confluent.Kafka</Title>
     <AssemblyName>Confluent.Kafka</AssemblyName>
-    <VersionPrefix>2.1.0-RC3</VersionPrefix>
+    <VersionPrefix>2.1.0</VersionPrefix>
     <TargetFrameworks>netstandard2.0;netstandard1.3;net462;net6.0</TargetFrameworks>
     <AllowUnsafeBlocks>true</AllowUnsafeBlocks>
     <GenerateDocumentationFile>true</GenerateDocumentationFile>
@@ -21,7 +21,7 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="librdkafka.redist" Version="2.1.0-RC3">
+    <PackageReference Include="librdkafka.redist" Version="2.1.0">
         <PrivateAssets Condition="'$(TargetFrameworkIdentifier)' == '.NETFramework'">None</PrivateAssets>
     </PackageReference>
     <PackageReference Include="System.Memory" Version="4.5.0" />
diff --git a/src/Confluent.Kafka/Consumer.cs b/src/Confluent.Kafka/Consumer.cs
@@ -794,6 +794,13 @@ public ConsumeResult<TKey, TValue> Consume(int millisecondsTimeout)
             try
             {
                 var msg = Util.Marshal.PtrToStructure<rd_kafka_message>(msgPtr);
+                int? msgLeaderEpoch = null;
+                Offset msgOffset = msg.offset;
+
+                if (msg.rkt != IntPtr.Zero && (msgOffset != Offset.Unset))
+                {
+                    msgLeaderEpoch = Librdkafka.message_leader_epoch(msgPtr);
+                }
 
                 string topic = null;
                 if (this.enableTopicNameMarshaling)
@@ -810,7 +817,7 @@ public ConsumeResult<TKey, TValue> Consume(int millisecondsTimeout)
                     {
                         TopicPartitionOffset = new TopicPartitionOffset(topic,
                             msg.partition, msg.offset,
-                            Librdkafka.message_leader_epoch(msgPtr)),
+                            msgLeaderEpoch),
                         Message = null,
                         IsPartitionEOF = true
                     };
@@ -857,7 +864,7 @@ public ConsumeResult<TKey, TValue> Consume(int millisecondsTimeout)
                         {
                             TopicPartitionOffset = new TopicPartitionOffset(topic,
                                 msg.partition, msg.offset,
-                                Librdkafka.message_leader_epoch(msgPtr)),
+                                msgLeaderEpoch),
                             Message = new Message<byte[], byte[]>
                             {
                                 Timestamp = timestamp,
@@ -890,7 +897,7 @@ public ConsumeResult<TKey, TValue> Consume(int millisecondsTimeout)
                         {
                             TopicPartitionOffset = new TopicPartitionOffset(topic,
                                 msg.partition, msg.offset,
-                                Librdkafka.message_leader_epoch(msgPtr)),
+                                msgLeaderEpoch),
                             Message = new Message<byte[], byte[]>
                             {
                                 Timestamp = timestamp,
@@ -924,7 +931,7 @@ public ConsumeResult<TKey, TValue> Consume(int millisecondsTimeout)
                         {
                             TopicPartitionOffset = new TopicPartitionOffset(topic,
                                 msg.partition, msg.offset,
-                                Librdkafka.message_leader_epoch(msgPtr)),
+                                msgLeaderEpoch),
                             Message = new Message<byte[], byte[]>
                             {
                                 Timestamp = timestamp,
@@ -942,7 +949,7 @@ public ConsumeResult<TKey, TValue> Consume(int millisecondsTimeout)
                 {
                     TopicPartitionOffset = new TopicPartitionOffset(topic,
                         msg.partition, msg.offset,
-                        Librdkafka.message_leader_epoch(msgPtr)),
+                        msgLeaderEpoch),
                     Message = new Message<TKey, TValue>
                     {
                         Timestamp = timestamp,
diff --git a/src/Confluent.Kafka/ErrorCode.cs b/src/Confluent.Kafka/ErrorCode.cs
@@ -329,7 +329,11 @@ public enum ErrorCode
         ///     No offset to automatically reset to
         /// </summary>
         Local_AutoOffsetReset = -140,
-
+        
+        /// <summary>
+        ///     Partition log truncation detected
+        /// </summary> 
+        Local_LogTruncation = -139,
 
         /// <summary>
         ///     Unknown broker error
diff --git a/src/Confluent.SchemaRegistry.Serdes.Avro/Confluent.SchemaRegistry.Serdes.Avro.csproj b/src/Confluent.SchemaRegistry.Serdes.Avro/Confluent.SchemaRegistry.Serdes.Avro.csproj
@@ -13,7 +13,7 @@
     <PackageId>Confluent.SchemaRegistry.Serdes.Avro</PackageId>
     <Title>Confluent.SchemaRegistry.Serdes.Avro</Title>
     <AssemblyName>Confluent.SchemaRegistry.Serdes.Avro</AssemblyName>
-    <VersionPrefix>2.1.0-RC3</VersionPrefix>
+    <VersionPrefix>2.1.0</VersionPrefix>
     <TargetFrameworks>netstandard2.0;</TargetFrameworks>
     <AllowUnsafeBlocks>true</AllowUnsafeBlocks>
     <GenerateDocumentationFile>true</GenerateDocumentationFile>
diff --git a/src/Confluent.SchemaRegistry.Serdes.Json/Confluent.SchemaRegistry.Serdes.Json.csproj b/src/Confluent.SchemaRegistry.Serdes.Json/Confluent.SchemaRegistry.Serdes.Json.csproj
@@ -13,7 +13,7 @@
     <PackageId>Confluent.SchemaRegistry.Serdes.Json</PackageId>
     <Title>Confluent.SchemaRegistry.Serdes.Json</Title>
     <AssemblyName>Confluent.SchemaRegistry.Serdes.Json</AssemblyName>
-    <VersionPrefix>2.1.0-RC3</VersionPrefix>
+    <VersionPrefix>2.1.0</VersionPrefix>
     <TargetFrameworks>netstandard2.0;</TargetFrameworks>
     <AllowUnsafeBlocks>true</AllowUnsafeBlocks>
     <GenerateDocumentationFile>true</GenerateDocumentationFile>
diff --git a/src/Confluent.SchemaRegistry.Serdes.Protobuf/Confluent.SchemaRegistry.Serdes.Protobuf.csproj b/src/Confluent.SchemaRegistry.Serdes.Protobuf/Confluent.SchemaRegistry.Serdes.Protobuf.csproj
@@ -13,7 +13,7 @@
     <PackageId>Confluent.SchemaRegistry.Serdes.Protobuf</PackageId>
     <Title>Confluent.SchemaRegistry.Serdes.Protobuf</Title>
     <AssemblyName>Confluent.SchemaRegistry.Serdes.Protobuf</AssemblyName>
-    <VersionPrefix>2.1.0-RC3</VersionPrefix>
+    <VersionPrefix>2.1.0</VersionPrefix>
     <TargetFrameworks>netstandard2.0;</TargetFrameworks>
     <AllowUnsafeBlocks>true</AllowUnsafeBlocks>
     <GenerateDocumentationFile>true</GenerateDocumentationFile>
diff --git a/src/Confluent.SchemaRegistry/Confluent.SchemaRegistry.csproj b/src/Confluent.SchemaRegistry/Confluent.SchemaRegistry.csproj
@@ -13,7 +13,7 @@
     <PackageId>Confluent.SchemaRegistry</PackageId>
     <Title>Confluent.SchemaRegistry</Title>
     <AssemblyName>Confluent.SchemaRegistry</AssemblyName>
-    <VersionPrefix>2.1.0-RC3</VersionPrefix>
+    <VersionPrefix>2.1.0</VersionPrefix>
     <TargetFrameworks>netstandard2.0;netstandard1.4</TargetFrameworks>
     <AllowUnsafeBlocks>true</AllowUnsafeBlocks>
     <GenerateDocumentationFile>true</GenerateDocumentationFile>

Original file line number	Diff line number	Diff line change
`@@ -1,4 +1,4 @@`
`1`		`-// * Auto-generated from librdkafka v2.1.0-RC3 * - do not modify manually.`
	`1`	`+// * Auto-generated from librdkafka v2.1.0 * - do not modify manually.`
`2`	`2`	`//`
`3`	`3`	`// Copyright 2018-2022 Confluent Inc.`
`4`	`4`	`//`
Original file line number	Diff line number	Diff line change
`@@ -794,6 +794,13 @@ public ConsumeResult<TKey, TValue> Consume(int millisecondsTimeout)`
`794`	`794`	`try`
`795`	`795`	`{`
`796`	`796`	`var msg = Util.Marshal.PtrToStructure<rd_kafka_message>(msgPtr);`
	`797`	`+ int? msgLeaderEpoch = null;`
	`798`	`+ Offset msgOffset = msg.offset;`
	`799`	`+`
	`800`	`+ if (msg.rkt != IntPtr.Zero && (msgOffset != Offset.Unset))`
	`801`	`+ {`
	`802`	`+ msgLeaderEpoch = Librdkafka.message_leader_epoch(msgPtr);`
	`803`	`+ }`
`797`	`804`
`798`	`805`	`string topic = null;`
`799`	`806`	`if (this.enableTopicNameMarshaling)`
`@@ -810,7 +817,7 @@ public ConsumeResult<TKey, TValue> Consume(int millisecondsTimeout)`
`810`	`817`	`{`
`811`	`818`	`TopicPartitionOffset = new TopicPartitionOffset(topic,`
`812`	`819`	`msg.partition, msg.offset,`
`813`		`- Librdkafka.message_leader_epoch(msgPtr)),`
	`820`	`+ msgLeaderEpoch),`
`814`	`821`	`Message = null,`
`815`	`822`	`IsPartitionEOF = true`
`816`	`823`	`};`
`@@ -857,7 +864,7 @@ public ConsumeResult<TKey, TValue> Consume(int millisecondsTimeout)`
`857`	`864`	`{`
`858`	`865`	`TopicPartitionOffset = new TopicPartitionOffset(topic,`
`859`	`866`	`msg.partition, msg.offset,`
`860`		`- Librdkafka.message_leader_epoch(msgPtr)),`
	`867`	`+ msgLeaderEpoch),`
`861`	`868`	`Message = new Message<byte[], byte[]>`
`862`	`869`	`{`
`863`	`870`	`Timestamp = timestamp,`
`@@ -890,7 +897,7 @@ public ConsumeResult<TKey, TValue> Consume(int millisecondsTimeout)`
`890`	`897`	`{`
`891`	`898`	`TopicPartitionOffset = new TopicPartitionOffset(topic,`
`892`	`899`	`msg.partition, msg.offset,`
`893`		`- Librdkafka.message_leader_epoch(msgPtr)),`
	`900`	`+ msgLeaderEpoch),`
`894`	`901`	`Message = new Message<byte[], byte[]>`
`895`	`902`	`{`
`896`	`903`	`Timestamp = timestamp,`
`@@ -924,7 +931,7 @@ public ConsumeResult<TKey, TValue> Consume(int millisecondsTimeout)`
`924`	`931`	`{`
`925`	`932`	`TopicPartitionOffset = new TopicPartitionOffset(topic,`
`926`	`933`	`msg.partition, msg.offset,`
`927`		`- Librdkafka.message_leader_epoch(msgPtr)),`
	`934`	`+ msgLeaderEpoch),`
`928`	`935`	`Message = new Message<byte[], byte[]>`
`929`	`936`	`{`
`930`	`937`	`Timestamp = timestamp,`
`@@ -942,7 +949,7 @@ public ConsumeResult<TKey, TValue> Consume(int millisecondsTimeout)`
`942`	`949`	`{`
`943`	`950`	`TopicPartitionOffset = new TopicPartitionOffset(topic,`
`944`	`951`	`msg.partition, msg.offset,`
`945`		`- Librdkafka.message_leader_epoch(msgPtr)),`
	`952`	`+ msgLeaderEpoch),`
`946`	`953`	`Message = new Message<TKey, TValue>`
`947`	`954`	`{`
`948`	`955`	`Timestamp = timestamp,`