Made message convenience properties on ConsumeResult obsolete (confluentinc#1215)

mhowlett · web-flow · commit 06581248f14c · 2020-03-16T13:02:09.000-07:00
diff --git a/examples/AdminClient/Program.cs b/examples/AdminClient/Program.cs
@@ -17,10 +17,7 @@
 // Refer to LICENSE for more information.
 
 using System;
-using System.Collections.Generic;
-using System.Linq;
 using System.Threading.Tasks;
-using Confluent.Kafka;
 using Confluent.Kafka.Admin;
 
 
diff --git a/examples/AvroBlogExamples/Program.cs b/examples/AvroBlogExamples/Program.cs
@@ -123,7 +123,7 @@ static void ConsumeSpecific(string bootstrapServers, string schemaRegistryUrl)
 
                             Console.WriteLine(
                                 consumeResult.Message.Timestamp.UtcDateTime.ToString("yyyy-MM-dd HH:mm:ss")
-                                + $": [{consumeResult.Value.Severity}] {consumeResult.Value.Message}");
+                                + $": [{consumeResult.Message.Value.Severity}] {consumeResult.Message.Value.Message}");
                         }
                         catch (ConsumeException e)
                         {
diff --git a/examples/AvroGeneric/Program.cs b/examples/AvroGeneric/Program.cs
@@ -76,7 +76,7 @@ static async Task Main(string[] args)
                             {
                                 var consumeResult = consumer.Consume(cts.Token);
 
-                                Console.WriteLine($"Key: {consumeResult.Message.Key}\nValue: {consumeResult.Value}");
+                                Console.WriteLine($"Key: {consumeResult.Message.Key}\nValue: {consumeResult.Message.Value}");
                             }
                             catch (ConsumeException e)
                             {
diff --git a/examples/AvroSpecific/Program.cs b/examples/AvroSpecific/Program.cs
@@ -84,7 +84,7 @@ static async Task Main(string[] args)
                             try
                             {
                                 var consumeResult = consumer.Consume(cts.Token);
-                                Console.WriteLine($"user name: {consumeResult.Message.Key}, favorite color: {consumeResult.Value.favorite_color}, hourly_rate: {consumeResult.Value.hourly_rate}");
+                                Console.WriteLine($"user name: {consumeResult.Message.Key}, favorite color: {consumeResult.Message.Value.favorite_color}, hourly_rate: {consumeResult.Message.Value.hourly_rate}");
                             }
                             catch (ConsumeException e)
                             {
diff --git a/examples/ConfluentCloud/Program.cs b/examples/ConfluentCloud/Program.cs
@@ -79,7 +79,7 @@ static void Main(string[] args)
                 try
                 {
                     var consumeResult = consumer.Consume();
-                    Console.WriteLine($"consumed: {consumeResult.Value}");
+                    Console.WriteLine($"consumed: {consumeResult.Message.Value}");
                 }
                 catch (ConsumeException e)
                 {
diff --git a/examples/Consumer/Program.cs b/examples/Consumer/Program.cs
@@ -92,7 +92,7 @@ public static void Run_Consume(string brokerList, List<string> topics, Cancellat
                                 continue;
                             }
 
-                            Console.WriteLine($"Received message at {consumeResult.TopicPartitionOffset}: {consumeResult.Value}");
+                            Console.WriteLine($"Received message at {consumeResult.TopicPartitionOffset}: {consumeResult.Message.Value}");
 
                             if (consumeResult.Offset % commitPeriod == 0)
                             {
@@ -163,7 +163,7 @@ public static void Run_ManualAssign(string brokerList, List<string> topics, Canc
                             // Note: End of partition notification has not been enabled, so
                             // it is guaranteed that the ConsumeResult instance corresponds
                             // to a Message, and not a PartitionEOF event.
-                            Console.WriteLine($"Received message at {consumeResult.TopicPartitionOffset}: ${consumeResult.Value}");
+                            Console.WriteLine($"Received message at {consumeResult.TopicPartitionOffset}: ${consumeResult.Message.Value}");
                         }
                         catch (ConsumeException e)
                         {
diff --git a/examples/Protobuf/Program.cs b/examples/Protobuf/Program.cs
@@ -81,7 +81,7 @@ static async Task Main(string[] args)
                             try
                             {
                                 var consumeResult = consumer.Consume(cts.Token);
-                                Console.WriteLine($"user name: {consumeResult.Message.Key}, favorite color: {consumeResult.Value.FavoriteColor}");
+                                Console.WriteLine($"user name: {consumeResult.Message.Key}, favorite color: {consumeResult.Message.Value.FavoriteColor}");
                             }
                             catch (ConsumeException e)
                             {
diff --git a/examples/Transactions/Program.cs b/examples/Transactions/Program.cs
@@ -323,7 +323,7 @@ static void Processor_MapWords(string brokerList, string clientId, CancellationT
 
                         producerState[cr.TopicPartition].Offset = cr.Offset;
 
-                        var words = Regex.Split(cr.Value.ToLower(), @"[^a-zA-Z_]").Where(s => s != String.Empty);
+                        var words = Regex.Split(cr.Message.Value.ToLower(), @"[^a-zA-Z_]").Where(s => s != String.Empty);
                         foreach (var w in words)
                         {
                             while (true)
@@ -444,7 +444,7 @@ public static void LoadCountState(RocksDb db, string brokerList, IEnumerable<Par
                     else
                     {
                         msgCount += 1;
-                        db.Put(Encoding.UTF8.GetBytes(cr.Key), BitConverter.GetBytes(cr.Value), columnFamily);
+                        db.Put(Encoding.UTF8.GetBytes(cr.Message.Key), BitConverter.GetBytes(cr.Message.Value), columnFamily);
                     }
                 }
             }
@@ -550,7 +550,7 @@ public static void Processor_AggregateWords(string brokerList, string clientId,
                         var cr = consumer.Consume(ct);
                         producerState[cr.TopicPartition].Offset = cr.Offset;
 
-                        var kBytes = Encoding.UTF8.GetBytes(cr.Key);
+                        var kBytes = Encoding.UTF8.GetBytes(cr.Message.Key);
                         var vBytes = db.Get(kBytes, columnFamily);
                         var v = vBytes == null ? 0 : BitConverter.ToInt32(vBytes);
                         var updatedV = v+1;
@@ -562,7 +562,7 @@ public static void Processor_AggregateWords(string brokerList, string clientId,
                             try
                             {
                                 producerState[cr.TopicPartition].Producer.Produce(
-                                    Topic_Counts, new Message<string, int> { Key = cr.Key, Value = updatedV });
+                                    Topic_Counts, new Message<string, int> { Key = cr.Message.Key, Value = updatedV });
                             }
                             catch (KafkaException e)
                             {
diff --git a/src/Confluent.Kafka/ConsumeResult.cs b/src/Confluent.Kafka/ConsumeResult.cs
@@ -16,6 +16,8 @@
 //
 // Refer to LICENSE for more information.
 
+using System;
+
 
 namespace Confluent.Kafka
 {
@@ -71,6 +73,7 @@ public TopicPartitionOffset TopicPartitionOffset
         /// <summary>
         ///     The Kafka message Key.
         /// </summary>
+        [Obsolete("Please access the message Key via .Message.Key.")]
         public TKey Key
         {
             get
@@ -87,6 +90,7 @@ public TKey Key
         /// <summary>
         ///     The Kafka message Value.
         /// </summary>
+        [Obsolete("Please access the message Value via .Message.Value.")]
         public TValue Value
         {
             get
@@ -103,6 +107,7 @@ public TValue Value
         /// <summary>
         ///     The Kafka message timestamp.
         /// </summary>
+        [Obsolete("Please access the message Timestamp via .Message.Timestamp.")]
         public Timestamp Timestamp
         {
             get
@@ -119,6 +124,7 @@ public Timestamp Timestamp
         /// <summary>
         ///     The Kafka message headers.
         /// </summary>
+        [Obsolete("Please access the message Headers via .Message.Headers.")]
         public Headers Headers
         {
             get
diff --git a/test/Confluent.Kafka.IntegrationTests/Tests/AssignOverloads.cs b/test/Confluent.Kafka.IntegrationTests/Tests/AssignOverloads.cs
@@ -63,7 +63,7 @@ public void AssignOverloads(string bootstrapServers)
                 consumer.Assign(new List<TopicPartitionOffset>() { new TopicPartitionOffset(dr.TopicPartition, dr.Offset) });
                 var cr = consumer.Consume(TimeSpan.FromSeconds(10));
                 consumer.Commit();
-                Assert.Equal(cr.Value, testString);
+                Assert.Equal(cr.Message.Value, testString);
                 
                 // Determine offset to consume from automatically.
                 consumer.Assign(new List<TopicPartition>() { dr.TopicPartition });
@@ -76,7 +76,7 @@ public void AssignOverloads(string bootstrapServers)
                 consumer.Assign(new TopicPartitionOffset(dr.TopicPartition, dr3.Offset));
                 cr = consumer.Consume(TimeSpan.FromSeconds(10));
                 consumer.Commit();
-                Assert.Equal(cr.Value, testString3);
+                Assert.Equal(cr.Message.Value, testString3);
 
                 // Determine offset to consume from automatically.
                 consumer.Assign(dr.TopicPartition);
diff --git a/test/Confluent.Kafka.IntegrationTests/Tests/Builder_CustomDefaults.cs b/test/Confluent.Kafka.IntegrationTests/Tests/Builder_CustomDefaults.cs
@@ -123,17 +123,17 @@ public void ProducerBuilder(string bootstrapServers)
             {
                 c.Assign(dr.TopicPartitionOffset);
                 var cr = c.Consume(TimeSpan.FromSeconds(10));
-                Assert.Equal("abc", cr.Key);
-                Assert.Equal("123", cr.Value);
+                Assert.Equal("abc", cr.Message.Key);
+                Assert.Equal("123", cr.Message.Value);
             }
 
             using (var c = new ConsumerBuilder<byte[], byte[]>(consumerConfig).Build())
             {
                 c.Assign(dr.TopicPartitionOffset);
                 var cr = c.Consume(TimeSpan.FromSeconds(10));
                 // check that each character is serialized into 4 bytes.
-                Assert.Equal(3*4, cr.Key.Length);
-                Assert.Equal(3*4, cr.Value.Length);
+                Assert.Equal(3*4, cr.Message.Key.Length);
+                Assert.Equal(3*4, cr.Message.Value.Length);
             }
 
             Assert.Equal(0, Library.HandleCount);
diff --git a/test/Confluent.Kafka.IntegrationTests/Tests/Consumer_DisableHeaders.cs b/test/Confluent.Kafka.IntegrationTests/Tests/Consumer_DisableHeaders.cs
@@ -64,8 +64,8 @@ public void Consumer_DisableHeaders(string bootstrapServers)
                 var record = consumer.Consume(TimeSpan.FromSeconds(10));
                 Assert.NotNull(record.Message);
                 Assert.Null(record.Message.Headers);
-                Assert.NotEqual(TimestampType.NotAvailable, record.Timestamp.Type);
-                Assert.NotEqual(0, record.Timestamp.UnixTimestampMs);
+                Assert.NotEqual(TimestampType.NotAvailable, record.Message.Timestamp.Type);
+                Assert.NotEqual(0, record.Message.Timestamp.UnixTimestampMs);
             }
 
             Assert.Equal(0, Library.HandleCount);
diff --git a/test/Confluent.Kafka.IntegrationTests/Tests/Consumer_DisableTimestamps.cs b/test/Confluent.Kafka.IntegrationTests/Tests/Consumer_DisableTimestamps.cs
@@ -64,8 +64,8 @@ public void Consumer_DisableTimestamps(string bootstrapServers)
                 var record = consumer.Consume(TimeSpan.FromSeconds(10));
                 Assert.NotNull(record.Message);
                 Assert.NotNull(record.Message.Headers);
-                Assert.Equal(TimestampType.NotAvailable, record.Timestamp.Type);
-                Assert.Equal(0, record.Timestamp.UnixTimestampMs);
+                Assert.Equal(TimestampType.NotAvailable, record.Message.Timestamp.Type);
+                Assert.Equal(0, record.Message.Timestamp.UnixTimestampMs);
             }
             
             Assert.Equal(0, Library.HandleCount);
diff --git a/test/Confluent.Kafka.IntegrationTests/Tests/Headers.cs b/test/Confluent.Kafka.IntegrationTests/Tests/Headers.cs
@@ -318,7 +318,7 @@ public void MessageHeaderProduceConsume(string bootstrapServers)
                 consumer.Assign(new TopicPartitionOffset(singlePartitionTopic, 0, nulldr.Offset));
                 var cr = consumer.Consume(TimeSpan.FromSeconds(10));
                 Assert.NotNull(cr?.Message);
-                Assert.Single(cr.Headers);
+                Assert.Single(cr.Message.Headers);
                 Assert.Equal("my-header", cr.Message.Headers[0].Key);
                 Assert.Null(cr.Message.Headers[0].GetValueBytes());
             }
diff --git a/test/Confluent.Kafka.IntegrationTests/Tests/Ignore.cs b/test/Confluent.Kafka.IntegrationTests/Tests/Ignore.cs
@@ -73,10 +73,10 @@ record = consumer.Consume(TimeSpan.FromSeconds(10));
 
                 ConsumeResult<Ignore, byte[]> record = consumer.Consume(TimeSpan.FromSeconds(10));
                 Assert.NotNull(record.Message);
-                Assert.Null(record.Key);
-                Assert.NotNull(record.Value);
-                Assert.Equal(42, record.Value[0]);
-                Assert.Equal(240, record.Value[1]);
+                Assert.Null(record.Message.Key);
+                Assert.NotNull(record.Message.Value);
+                Assert.Equal(42, record.Message.Value[0]);
+                Assert.Equal(240, record.Message.Value[1]);
             }
 
             Assert.Equal(0, Library.HandleCount);
diff --git a/test/Confluent.Kafka.IntegrationTests/Tests/Producer_Handles.cs b/test/Confluent.Kafka.IntegrationTests/Tests/Producer_Handles.cs
@@ -87,62 +87,62 @@ public void Producer_Handles(string bootstrapServers)
                 {
                     consumer.Assign(new TopicPartitionOffset(topic.Name, 0, 0));
                     var r1 = consumer.Consume(TimeSpan.FromSeconds(10));
-                    Assert.Equal(new byte[] { 42 }, r1.Key);
-                    Assert.Equal(new byte[] { 33 }, r1.Value);
+                    Assert.Equal(new byte[] { 42 }, r1.Message.Key);
+                    Assert.Equal(new byte[] { 33 }, r1.Message.Value);
                     Assert.Equal(0, r1.Offset);
                 }
 
                 using (var consumer = new ConsumerBuilder<string, string>(consumerConfig).Build())
                 {
                     consumer.Assign(new TopicPartitionOffset(topic.Name, 0, 1));
                     var r2 = consumer.Consume(TimeSpan.FromSeconds(10));
-                    Assert.Equal("hello", r2.Key);
-                    Assert.Equal("world", r2.Value);
+                    Assert.Equal("hello", r2.Message.Key);
+                    Assert.Equal("world", r2.Message.Value);
                     Assert.Equal(1, r2.Offset);
                 }
 
                 using (var consumer = new ConsumerBuilder<byte[], byte[]>(consumerConfig).Build())
                 {
                     consumer.Assign(new TopicPartitionOffset(topic.Name, 0, 2));
                     var r3 = consumer.Consume(TimeSpan.FromSeconds(10));
-                    Assert.Equal(new byte[] { 40 }, r3.Key);
-                    Assert.Equal(new byte[] { 31 }, r3.Value);
+                    Assert.Equal(new byte[] { 40 }, r3.Message.Key);
+                    Assert.Equal(new byte[] { 31 }, r3.Message.Value);
                     Assert.Equal(2, r3.Offset);
                 }
 
                 using (var consumer = new ConsumerBuilder<int, string>(consumerConfig).Build())
                 {
                     consumer.Assign(new TopicPartitionOffset(topic.Name, 0, 3));
                     var r4 = consumer.Consume(TimeSpan.FromSeconds(10));
-                    Assert.Equal(42, r4.Key);
-                    Assert.Equal("mellow world", r4.Value);
+                    Assert.Equal(42, r4.Message.Key);
+                    Assert.Equal("mellow world", r4.Message.Value);
                     Assert.Equal(3, r4.Offset);
                 }
 
                 using (var consumer = new ConsumerBuilder<int, int>(consumerConfig).Build())
                 {
                     consumer.Assign(new TopicPartitionOffset(topic.Name, 0, 4));
                     var r5 = consumer.Consume(TimeSpan.FromSeconds(10));
-                    Assert.Equal(int.MaxValue, r5.Key);
-                    Assert.Equal(int.MinValue, r5.Value);
+                    Assert.Equal(int.MaxValue, r5.Message.Key);
+                    Assert.Equal(int.MinValue, r5.Message.Value);
                     Assert.Equal(4, r5.Offset);
                 }
 
                 using (var consumer = new ConsumerBuilder<string, byte[]>(consumerConfig).Build())
                 {
                     consumer.Assign(new TopicPartitionOffset(topic.Name, 0, 5));
                     var r6 = consumer.Consume(TimeSpan.FromSeconds(10));
-                    Assert.Equal("yellow mould", r6.Key);
-                    Assert.Equal(new byte[] { 69 }, r6.Value);
+                    Assert.Equal("yellow mould", r6.Message.Key);
+                    Assert.Equal(new byte[] { 69 }, r6.Message.Value);
                     Assert.Equal(5, r6.Offset);
                 }
 
                 using (var consumer = new ConsumerBuilder<double, double>(consumerConfig).Build())
                 {
                     consumer.Assign(new TopicPartitionOffset(topic.Name, 0, 6));
                     var r7 = consumer.Consume(TimeSpan.FromSeconds(10));
-                    Assert.Equal(44.0, r7.Key);
-                    Assert.Equal(234.4, r7.Value);
+                    Assert.Equal(44.0, r7.Message.Key);
+                    Assert.Equal(234.4, r7.Message.Value);
                     Assert.Equal(6, r7.Offset);
                 }
             }
diff --git a/test/Confluent.Kafka.Transactions/TestConsumer.cs b/test/Confluent.Kafka.Transactions/TestConsumer.cs
@@ -39,11 +39,11 @@ public void Run()
                 {
                     var cr = consumer.Consume();
 
-                    if (!lasts.ContainsKey(cr.Key)) { lasts.Add(cr.Key, -1); }
-                    if (cr.Value == lasts[cr.Key] + 1) { Console.Write("."); }
-                    else { Console.Write($"[producer {cr.Key} expected seq {lasts[cr.Key]+1} but got {cr.Value}]"); break; }
+                    if (!lasts.ContainsKey(cr.Message.Key)) { lasts.Add(cr.Message.Key, -1); }
+                    if (cr.Message.Value == lasts[cr.Message.Key] + 1) { Console.Write("."); }
+                    else { Console.Write($"[producer {cr.Message.Key} expected seq {lasts[cr.Message.Key]+1} but got {cr.Message.Value}]"); break; }
                     Console.Out.Flush();
-                    lasts[cr.Key] = cr.Value;
+                    lasts[cr.Message.Key] = cr.Message.Value;
                 }
             }
             catch (Exception e)
diff --git a/test/Confluent.SchemaRegistry.Serdes.IntegrationTests/Tests_Avro/AvroAndRegular.cs b/test/Confluent.SchemaRegistry.Serdes.IntegrationTests/Tests_Avro/AvroAndRegular.cs
@@ -112,8 +112,8 @@ public static void AvoAndRegular(string bootstrapServers, string schemaRegistryS
                     {
                         consumer.Assign(new TopicPartitionOffset(topic1.Name, 0, 0));
                         var cr = consumer.Consume();
-                        Assert.Equal("hello", cr.Key);
-                        Assert.Equal("world", cr.Value);
+                        Assert.Equal("hello", cr.Message.Key);
+                        Assert.Equal("world", cr.Message.Value);
                     }
 
                     using (var consumer =
@@ -123,8 +123,8 @@ public static void AvoAndRegular(string bootstrapServers, string schemaRegistryS
                     {
                         consumer.Assign(new TopicPartitionOffset(topic2.Name, 0, 0));
                         var cr = consumer.Consume();
-                        Assert.Equal("hello", cr.Key);
-                        Assert.Equal("world", cr.Value);
+                        Assert.Equal("hello", cr.Message.Key);
+                        Assert.Equal("world", cr.Message.Value);
                     }
 
                     using (var consumer =
diff --git a/test/Confluent.SchemaRegistry.Serdes.IntegrationTests/Tests_Avro/PrimitiveTypes.cs b/test/Confluent.SchemaRegistry.Serdes.IntegrationTests/Tests_Avro/PrimitiveTypes.cs
@@ -244,8 +244,8 @@ public static void PrimitiveTypes(string bootstrapServers, string schemaRegistry
                 {
                     consumer.Assign(new List<TopicPartitionOffset> { new TopicPartitionOffset(nullTopic, 0, 0) });
                     var result8 = consumer.Consume(TimeSpan.FromSeconds(10));
-                    Assert.Null(result8.Key);
-                    Assert.Null(result8.Value);
+                    Assert.Null(result8.Message.Key);
+                    Assert.Null(result8.Message.Value);
                 }
             }
         }
diff --git a/test/Confluent.SchemaRegistry.Serdes.UnitTests/Configuration.cs b/test/Confluent.SchemaRegistry.Serdes.UnitTests/Configuration.cs
@@ -33,7 +33,7 @@ public ConfigurationTests()
         {
             testTopic = "topic";
             var schemaRegistryMock = new Mock<ISchemaRegistryClient>();
-            schemaRegistryMock.Setup(x => x.RegisterSchemaAsync(testTopic + "-value", It.IsAny<string>())).ReturnsAsync(
+            schemaRegistryMock.Setup(x => x.RegisterSchemaAsync(testTopic + "-value", It.IsAny<Schema>())).ReturnsAsync(
                 (string topic, string schema) => store.TryGetValue(schema, out int id) ? id : store[schema] = store.Count + 1
             );
             schemaRegistryMock.Setup(x => x.GetSchemaAsync(It.IsAny<int>(), It.IsAny<string>())).ReturnsAsync(

Original file line number	Diff line number	Diff line change
`@@ -123,7 +123,7 @@ static void ConsumeSpecific(string bootstrapServers, string schemaRegistryUrl)`
`123`	`123`
`124`	`124`	`Console.WriteLine(`
`125`	`125`	`consumeResult.Message.Timestamp.UtcDateTime.ToString("yyyy-MM-dd HH:mm:ss")`
`126`		`- + $": [{consumeResult.Value.Severity}] {consumeResult.Value.Message}");`
	`126`	`+ + $": [{consumeResult.Message.Value.Severity}] {consumeResult.Message.Value.Message}");`
`127`	`127`	`}`
`128`	`128`	`catch (ConsumeException e)`
`129`	`129`	`{`
Original file line number	Diff line number	Diff line change
`@@ -76,7 +76,7 @@ static async Task Main(string[] args)`
`76`	`76`	`{`
`77`	`77`	`var consumeResult = consumer.Consume(cts.Token);`
`78`	`78`
`79`		`- Console.WriteLine($"Key: {consumeResult.Message.Key}\nValue: {consumeResult.Value}");`
	`79`	`+ Console.WriteLine($"Key: {consumeResult.Message.Key}\nValue: {consumeResult.Message.Value}");`
`80`	`80`	`}`
`81`	`81`	`catch (ConsumeException e)`
`82`	`82`	`{`
Original file line number	Diff line number	Diff line change
`@@ -84,7 +84,7 @@ static async Task Main(string[] args)`
`84`	`84`	`try`
`85`	`85`	`{`
`86`	`86`	`var consumeResult = consumer.Consume(cts.Token);`
`87`		`- Console.WriteLine($"user name: {consumeResult.Message.Key}, favorite color: {consumeResult.Value.favorite_color}, hourly_rate: {consumeResult.Value.hourly_rate}");`
	`87`	`+ Console.WriteLine($"user name: {consumeResult.Message.Key}, favorite color: {consumeResult.Message.Value.favorite_color}, hourly_rate: {consumeResult.Message.Value.hourly_rate}");`
`88`	`88`	`}`
`89`	`89`	`catch (ConsumeException e)`
`90`	`90`	`{`
Original file line number	Diff line number	Diff line change
`@@ -79,7 +79,7 @@ static void Main(string[] args)`
`79`	`79`	`try`
`80`	`80`	`{`
`81`	`81`	`var consumeResult = consumer.Consume();`
`82`		`- Console.WriteLine($"consumed: {consumeResult.Value}");`
	`82`	`+ Console.WriteLine($"consumed: {consumeResult.Message.Value}");`
`83`	`83`	`}`
`84`	`84`	`catch (ConsumeException e)`
`85`	`85`	`{`
Original file line number	Diff line number	Diff line change
`@@ -92,7 +92,7 @@ public static void Run_Consume(string brokerList, List<string> topics, Cancellat`
`92`	`92`	`continue;`
`93`	`93`	`}`
`94`	`94`
`95`		`- Console.WriteLine($"Received message at {consumeResult.TopicPartitionOffset}: {consumeResult.Value}");`
	`95`	`+ Console.WriteLine($"Received message at {consumeResult.TopicPartitionOffset}: {consumeResult.Message.Value}");`
`96`	`96`
`97`	`97`	`if (consumeResult.Offset % commitPeriod == 0)`
`98`	`98`	`{`
`@@ -163,7 +163,7 @@ public static void Run_ManualAssign(string brokerList, List<string> topics, Canc`
`163`	`163`	`// Note: End of partition notification has not been enabled, so`
`164`	`164`	`// it is guaranteed that the ConsumeResult instance corresponds`
`165`	`165`	`// to a Message, and not a PartitionEOF event.`
`166`		`- Console.WriteLine($"Received message at {consumeResult.TopicPartitionOffset}: ${consumeResult.Value}");`
	`166`	`+ Console.WriteLine($"Received message at {consumeResult.TopicPartitionOffset}: ${consumeResult.Message.Value}");`
`167`	`167`	`}`
`168`	`168`	`catch (ConsumeException e)`
`169`	`169`	`{`
Original file line number	Diff line number	Diff line change
`@@ -81,7 +81,7 @@ static async Task Main(string[] args)`
`81`	`81`	`try`
`82`	`82`	`{`
`83`	`83`	`var consumeResult = consumer.Consume(cts.Token);`
`84`		`- Console.WriteLine($"user name: {consumeResult.Message.Key}, favorite color: {consumeResult.Value.FavoriteColor}");`
	`84`	`+ Console.WriteLine($"user name: {consumeResult.Message.Key}, favorite color: {consumeResult.Message.Value.FavoriteColor}");`
`85`	`85`	`}`
`86`	`86`	`catch (ConsumeException e)`
`87`	`87`	`{`
Original file line number	Diff line number	Diff line change
`@@ -323,7 +323,7 @@ static void Processor_MapWords(string brokerList, string clientId, CancellationT`
`323`	`323`
`324`	`324`	`producerState[cr.TopicPartition].Offset = cr.Offset;`
`325`	`325`
`326`		`- var words = Regex.Split(cr.Value.ToLower(), @"[^a-zA-Z_]").Where(s => s != String.Empty);`
	`326`	`+ var words = Regex.Split(cr.Message.Value.ToLower(), @"[^a-zA-Z_]").Where(s => s != String.Empty);`
`327`	`327`	`foreach (var w in words)`
`328`	`328`	`{`
`329`	`329`	`while (true)`
`@@ -444,7 +444,7 @@ public static void LoadCountState(RocksDb db, string brokerList, IEnumerable<Par`
`444`	`444`	`else`
`445`	`445`	`{`
`446`	`446`	`msgCount += 1;`
`447`		`- db.Put(Encoding.UTF8.GetBytes(cr.Key), BitConverter.GetBytes(cr.Value), columnFamily);`
	`447`	`+ db.Put(Encoding.UTF8.GetBytes(cr.Message.Key), BitConverter.GetBytes(cr.Message.Value), columnFamily);`
`448`	`448`	`}`
`449`	`449`	`}`
`450`	`450`	`}`
`@@ -550,7 +550,7 @@ public static void Processor_AggregateWords(string brokerList, string clientId,`
`550`	`550`	`var cr = consumer.Consume(ct);`
`551`	`551`	`producerState[cr.TopicPartition].Offset = cr.Offset;`
`552`	`552`
`553`		`- var kBytes = Encoding.UTF8.GetBytes(cr.Key);`
	`553`	`+ var kBytes = Encoding.UTF8.GetBytes(cr.Message.Key);`
`554`	`554`	`var vBytes = db.Get(kBytes, columnFamily);`
`555`	`555`	`var v = vBytes == null ? 0 : BitConverter.ToInt32(vBytes);`
`556`	`556`	`var updatedV = v+1;`
`@@ -562,7 +562,7 @@ public static void Processor_AggregateWords(string brokerList, string clientId,`
`562`	`562`	`try`
`563`	`563`	`{`
`564`	`564`	`producerState[cr.TopicPartition].Producer.Produce(`
`565`		`- Topic_Counts, new Message<string, int> { Key = cr.Key, Value = updatedV });`
	`565`	`+ Topic_Counts, new Message<string, int> { Key = cr.Message.Key, Value = updatedV });`
`566`	`566`	`}`
`567`	`567`	`catch (KafkaException e)`
`568`	`568`	`{`
Original file line number	Diff line number	Diff line change
`@@ -123,17 +123,17 @@ public void ProducerBuilder(string bootstrapServers)`
`123`	`123`	`{`
`124`	`124`	`c.Assign(dr.TopicPartitionOffset);`
`125`	`125`	`var cr = c.Consume(TimeSpan.FromSeconds(10));`
`126`		`- Assert.Equal("abc", cr.Key);`
`127`		`- Assert.Equal("123", cr.Value);`
	`126`	`+ Assert.Equal("abc", cr.Message.Key);`
	`127`	`+ Assert.Equal("123", cr.Message.Value);`
`128`	`128`	`}`
`129`	`129`
`130`	`130`	`using (var c = new ConsumerBuilder<byte[], byte[]>(consumerConfig).Build())`
`131`	`131`	`{`
`132`	`132`	`c.Assign(dr.TopicPartitionOffset);`
`133`	`133`	`var cr = c.Consume(TimeSpan.FromSeconds(10));`
`134`	`134`	`// check that each character is serialized into 4 bytes.`
`135`		`- Assert.Equal(3*4, cr.Key.Length);`
`136`		`- Assert.Equal(3*4, cr.Value.Length);`
	`135`	`+ Assert.Equal(3*4, cr.Message.Key.Length);`
	`136`	`+ Assert.Equal(3*4, cr.Message.Value.Length);`
`137`	`137`	`}`
`138`	`138`
`139`	`139`	`Assert.Equal(0, Library.HandleCount);`
Original file line number	Diff line number	Diff line change
`@@ -318,7 +318,7 @@ public void MessageHeaderProduceConsume(string bootstrapServers)`
`318`	`318`	`consumer.Assign(new TopicPartitionOffset(singlePartitionTopic, 0, nulldr.Offset));`
`319`	`319`	`var cr = consumer.Consume(TimeSpan.FromSeconds(10));`
`320`	`320`	`Assert.NotNull(cr?.Message);`
`321`		`- Assert.Single(cr.Headers);`
	`321`	`+ Assert.Single(cr.Message.Headers);`
`322`	`322`	`Assert.Equal("my-header", cr.Message.Headers[0].Key);`
`323`	`323`	`Assert.Null(cr.Message.Headers[0].GetValueBytes());`
`324`	`324`	`}`
Original file line number	Diff line number	Diff line change
`@@ -87,62 +87,62 @@ public void Producer_Handles(string bootstrapServers)`
`87`	`87`	`{`
`88`	`88`	`consumer.Assign(new TopicPartitionOffset(topic.Name, 0, 0));`
`89`	`89`	`var r1 = consumer.Consume(TimeSpan.FromSeconds(10));`
`90`		`- Assert.Equal(new byte[] { 42 }, r1.Key);`
`91`		`- Assert.Equal(new byte[] { 33 }, r1.Value);`
	`90`	`+ Assert.Equal(new byte[] { 42 }, r1.Message.Key);`
	`91`	`+ Assert.Equal(new byte[] { 33 }, r1.Message.Value);`
`92`	`92`	`Assert.Equal(0, r1.Offset);`
`93`	`93`	`}`
`94`	`94`
`95`	`95`	`using (var consumer = new ConsumerBuilder<string, string>(consumerConfig).Build())`
`96`	`96`	`{`
`97`	`97`	`consumer.Assign(new TopicPartitionOffset(topic.Name, 0, 1));`
`98`	`98`	`var r2 = consumer.Consume(TimeSpan.FromSeconds(10));`
`99`		`- Assert.Equal("hello", r2.Key);`
`100`		`- Assert.Equal("world", r2.Value);`
	`99`	`+ Assert.Equal("hello", r2.Message.Key);`
	`100`	`+ Assert.Equal("world", r2.Message.Value);`
`101`	`101`	`Assert.Equal(1, r2.Offset);`
`102`	`102`	`}`
`103`	`103`
`104`	`104`	`using (var consumer = new ConsumerBuilder<byte[], byte[]>(consumerConfig).Build())`
`105`	`105`	`{`
`106`	`106`	`consumer.Assign(new TopicPartitionOffset(topic.Name, 0, 2));`
`107`	`107`	`var r3 = consumer.Consume(TimeSpan.FromSeconds(10));`
`108`		`- Assert.Equal(new byte[] { 40 }, r3.Key);`
`109`		`- Assert.Equal(new byte[] { 31 }, r3.Value);`
	`108`	`+ Assert.Equal(new byte[] { 40 }, r3.Message.Key);`
	`109`	`+ Assert.Equal(new byte[] { 31 }, r3.Message.Value);`
`110`	`110`	`Assert.Equal(2, r3.Offset);`
`111`	`111`	`}`
`112`	`112`
`113`	`113`	`using (var consumer = new ConsumerBuilder<int, string>(consumerConfig).Build())`
`114`	`114`	`{`
`115`	`115`	`consumer.Assign(new TopicPartitionOffset(topic.Name, 0, 3));`
`116`	`116`	`var r4 = consumer.Consume(TimeSpan.FromSeconds(10));`
`117`		`- Assert.Equal(42, r4.Key);`
`118`		`- Assert.Equal("mellow world", r4.Value);`
	`117`	`+ Assert.Equal(42, r4.Message.Key);`
	`118`	`+ Assert.Equal("mellow world", r4.Message.Value);`
`119`	`119`	`Assert.Equal(3, r4.Offset);`
`120`	`120`	`}`
`121`	`121`
`122`	`122`	`using (var consumer = new ConsumerBuilder<int, int>(consumerConfig).Build())`
`123`	`123`	`{`
`124`	`124`	`consumer.Assign(new TopicPartitionOffset(topic.Name, 0, 4));`
`125`	`125`	`var r5 = consumer.Consume(TimeSpan.FromSeconds(10));`
`126`		`- Assert.Equal(int.MaxValue, r5.Key);`
`127`		`- Assert.Equal(int.MinValue, r5.Value);`
	`126`	`+ Assert.Equal(int.MaxValue, r5.Message.Key);`
	`127`	`+ Assert.Equal(int.MinValue, r5.Message.Value);`
`128`	`128`	`Assert.Equal(4, r5.Offset);`
`129`	`129`	`}`
`130`	`130`
`131`	`131`	`using (var consumer = new ConsumerBuilder<string, byte[]>(consumerConfig).Build())`
`132`	`132`	`{`
`133`	`133`	`consumer.Assign(new TopicPartitionOffset(topic.Name, 0, 5));`
`134`	`134`	`var r6 = consumer.Consume(TimeSpan.FromSeconds(10));`
`135`		`- Assert.Equal("yellow mould", r6.Key);`
`136`		`- Assert.Equal(new byte[] { 69 }, r6.Value);`
	`135`	`+ Assert.Equal("yellow mould", r6.Message.Key);`
	`136`	`+ Assert.Equal(new byte[] { 69 }, r6.Message.Value);`
`137`	`137`	`Assert.Equal(5, r6.Offset);`
`138`	`138`	`}`
`139`	`139`
`140`	`140`	`using (var consumer = new ConsumerBuilder<double, double>(consumerConfig).Build())`
`141`	`141`	`{`
`142`	`142`	`consumer.Assign(new TopicPartitionOffset(topic.Name, 0, 6));`
`143`	`143`	`var r7 = consumer.Consume(TimeSpan.FromSeconds(10));`
`144`		`- Assert.Equal(44.0, r7.Key);`
`145`		`- Assert.Equal(234.4, r7.Value);`
	`144`	`+ Assert.Equal(44.0, r7.Message.Key);`
	`145`	`+ Assert.Equal(234.4, r7.Message.Value);`
`146`	`146`	`Assert.Equal(6, r7.Offset);`
`147`	`147`	`}`
`148`	`148`	`}`