PersistenceStatus -> Status + integration tests (#807)

Author: mhowlett

src/Confluent.Kafka/DeliveryResult.cs

@@ -64,7 +64,7 @@ public TopicPartitionOffset TopicPartitionOffset
         /// <summary>
         ///     The persistence status of the message
         /// </summary>
-        public PersistenceStatus PersistenceStatus { get; set; }
+        public PersistenceStatus Status { get; set; }
 
         /// <summary>
         ///     The Kafka message.

src/Confluent.Kafka/Producer.cs

@@ -196,7 +196,7 @@ private void DeliveryReportCallbackImpl(IntPtr rk, IntPtr rkmessage, IntPtr opaq
                     Partition = msg.partition, 
                     Offset = msg.offset, 
                     Error = KafkaHandle.CreatePossiblyFatalError(msg.err, null),
-                    PersistenceStatus = messageStatus,
+                    Status = messageStatus,
                     Message = new Message<Null, Null> { Timestamp = new Timestamp(timestamp, (TimestampType)timestampType), Headers = headers }
                 }
             );
@@ -932,7 +932,7 @@ public void HandleDeliveryReport(DeliveryReport<Null, Null> deliveryReport)
                 var dr = new DeliveryResult<K, V>
                 {
                     TopicPartitionOffset = deliveryReport.TopicPartitionOffset,
-                    PersistenceStatus = deliveryReport.PersistenceStatus,
+                    Status = deliveryReport.Status,
                     Message = new Message<K, V>
                     {
                         Key = Key,
@@ -995,7 +995,7 @@ public void HandleDeliveryReport(DeliveryReport<Null, Null> deliveryReport)
                 var dr = new DeliveryReport<K, V>
                 {
                     TopicPartitionOffsetError = deliveryReport.TopicPartitionOffsetError,
-                    PersistenceStatus = deliveryReport.PersistenceStatus,
+                    Status = deliveryReport.Status,
                     Message = new Message<K, V> 
                     {
                         Key = Key,

test/Confluent.Kafka.IntegrationTests/Tests/Producer_BeginProduce.cs

@@ -46,6 +46,7 @@ public void Producer_BeginProduce(string bootstrapServers)
             Action<DeliveryReport<string, string>> dh = (DeliveryReport<string, string> dr) =>
             {
                 Assert.Equal(ErrorCode.NoError, dr.Error.Code);
+                Assert.Equal(PersistenceStatus.Persisted, dr.Status);
                 Assert.Equal((Partition)0, dr.Partition);
                 Assert.Equal(singlePartitionTopic, dr.Topic);
                 Assert.True(dr.Offset >= 0);
@@ -78,6 +79,7 @@ public void Producer_BeginProduce(string bootstrapServers)
             Action<DeliveryReport<byte[], byte[]>> dh2 = (DeliveryReport<byte[], byte[]> dr) =>
             {
                 Assert.Equal(ErrorCode.NoError, dr.Error.Code);
+                Assert.Equal(PersistenceStatus.Persisted, dr.Status);
                 Assert.Equal((Partition)0, dr.Partition);
                 Assert.Equal(singlePartitionTopic, dr.Topic);
                 Assert.True(dr.Offset >= 0);

test/Confluent.Kafka.IntegrationTests/Tests/Producer_BeginProduce_Error.cs

@@ -48,7 +48,7 @@ public void Producer_BeginProduce_Error(string bootstrapServers)
                 Assert.Equal(Offset.Invalid, dr.Offset);
                 Assert.Null(dr.Message.Key);
                 Assert.Equal("test", dr.Message.Value);
-                Assert.Equal(PersistenceStatus.NotPersisted, dr.PersistenceStatus);
+                Assert.Equal(PersistenceStatus.NotPersisted, dr.Status);
                 Assert.Equal(TimestampType.NotAvailable, dr.Message.Timestamp.Type);
                 count += 1;
             };

test/Confluent.Kafka.IntegrationTests/Tests/Producer_BeginProduce_Null.cs

@@ -41,6 +41,7 @@ public void Producer_BeginProduce_Null(string bootstrapServers)
             Action<DeliveryReport<Null, Null>> dh = (DeliveryReport<Null, Null> dr) => 
             {
                 Assert.Equal(ErrorCode.NoError, dr.Error.Code);
+                Assert.Equal(PersistenceStatus.Persisted, dr.Status);
                 Assert.False(dr.Error.IsFatal);
                 Assert.Equal((Partition)0, dr.Partition);
                 Assert.Equal(singlePartitionTopic, dr.Topic);

test/Confluent.Kafka.IntegrationTests/Tests/Producer_ProduceAsync_Error.cs

@@ -61,6 +61,7 @@ public void Producer_ProduceAsync_Error(string bootstrapServers)
                 var err = ((ProduceException<string, string>)inner).Error;
 
                 Assert.True(err.IsError);
+                Assert.Equal(PersistenceStatus.NotPersisted, dr.Status);
                 Assert.False(err.IsFatal);
                 Assert.Equal(partitionedTopic, dr.Topic);
                 Assert.Equal(Offset.Invalid, dr.Offset);

test/Confluent.Kafka.IntegrationTests/Tests/Producer_ProduceAsync_Task.cs

@@ -56,6 +56,7 @@ public void Producer_ProduceAsync_Task(string bootstrapServers)
             for (int i=0; i<2; ++i)
             {
                 var dr = drs[i].Result;
+                Assert.Equal(PersistenceStatus.Persisted, dr.Status);
                 Assert.Equal(partitionedTopic, dr.Topic);
                 Assert.True(dr.Offset >= 0);
                 Assert.True(dr.Partition == 0 || dr.Partition == 1);

test/Confluent.SchemaRegistry.Serdes.IntegrationTests/Tests/AutoRegisterSchemaDisabled.cs

@@ -82,7 +82,7 @@ public static void AutoRegisterSchemaDisabled(string bootstrapServers, string sc
                             Assert.Equal(Offset.Invalid, ((ProduceException<string, int>)e).DeliveryResult.Offset);
                             Assert.Equal(Partition.Any, ((ProduceException<string, int>)e).DeliveryResult.Partition);
                             Assert.Equal(guidTopic, ((ProduceException<string, int>)e).DeliveryResult.Topic);
-                            Assert.Equal(PersistenceStatus.NotPersisted, ((ProduceException<string, int>)e).DeliveryResult.PersistenceStatus);
+                            Assert.Equal(PersistenceStatus.NotPersisted, ((ProduceException<string, int>)e).DeliveryResult.Status);
                             Assert.Equal(Timestamp.Default, ((ProduceException<string, int>)e).DeliveryResult.Timestamp);
                             Assert.Null(((ProduceException<string, int>)e).DeliveryResult.Headers);