Improve tests

confluentinc · Nov 18, 2024 · 79bb726 · 79bb726
1 parent 6940abe
commit 79bb726
Show file tree

Hide file tree

Showing 7 changed files with 228 additions and 31 deletions.
diff --git a/test/Confluent.Kafka.IntegrationTests/Tests/Producer_Produce.cs b/test/Confluent.Kafka.IntegrationTests/Tests/Producer_Produce.cs
@@ -107,6 +107,71 @@ public void Producer_Produce(string bootstrapServers)
 
             Assert.Equal(2, count);
 
+            // Memory<byte> case.
+
+            count = 0;
+            Action<DeliveryReport<Memory<byte>, ReadOnlyMemory<byte>>> dh3 = 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);
+                Assert.Equal($"test key {count + 42}", Encoding.UTF8.GetString(dr.Message.Key.Span));
+                Assert.Equal($"test val {count + 42}", Encoding.UTF8.GetString(dr.Message.Value.Span));
+                Assert.Equal(TimestampType.CreateTime, dr.Message.Timestamp.Type);
+                Assert.True(Math.Abs((DateTime.UtcNow - dr.Message.Timestamp.UtcDateTime).TotalMinutes) < 1.0);
+                count += 1;
+            };
+
+            using (var producer = new TestProducerBuilder<Memory<byte>, ReadOnlyMemory<byte>>(producerConfig).Build())
+            {
+                producer.Produce(
+                    new TopicPartition(singlePartitionTopic, 0),
+                    new Message<Memory<byte>, ReadOnlyMemory<byte>> { Key = Encoding.UTF8.GetBytes("test key 42"), Value = Encoding.UTF8.GetBytes("test val 42") }, dh3);
+
+                producer.Produce(
+                    singlePartitionTopic,
+                    new Message<Memory<byte>, ReadOnlyMemory<byte>> { Key = Encoding.UTF8.GetBytes("test key 43"), Value = Encoding.UTF8.GetBytes("test val 43") }, dh3);
+
+                producer.Flush(TimeSpan.FromSeconds(10));
+            }
+
+            Assert.Equal(2, count);
+
+            // Memory<byte>? case.
+
+            count = 0;
+            Action<DeliveryReport<ReadOnlyMemory<byte>?, Memory<byte>?>> dh4 = 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);
+                Assert.True(dr.Message.Key.HasValue);
+                Assert.Equal($"test key {count + 42}", Encoding.UTF8.GetString(dr.Message.Key.Value.Span));
+                Assert.True(dr.Message.Value.HasValue);
+                Assert.Equal($"test val {count + 42}", Encoding.UTF8.GetString(dr.Message.Value.Value.Span));
+                Assert.Equal(TimestampType.CreateTime, dr.Message.Timestamp.Type);
+                Assert.True(Math.Abs((DateTime.UtcNow - dr.Message.Timestamp.UtcDateTime).TotalMinutes) < 1.0);
+                count += 1;
+            };
+
+            using (var producer = new TestProducerBuilder<ReadOnlyMemory<byte>?, Memory<byte>?>(producerConfig).Build())
+            {
+                producer.Produce(
+                    new TopicPartition(singlePartitionTopic, 0),
+                    new Message<ReadOnlyMemory<byte>?, Memory<byte>?> { Key = Encoding.UTF8.GetBytes("test key 42"), Value = Encoding.UTF8.GetBytes("test val 42") }, dh4);
+
+                producer.Produce(
+                    singlePartitionTopic,
+                    new Message<ReadOnlyMemory<byte>?, Memory<byte>?> { Key = Encoding.UTF8.GetBytes("test key 43"), Value = Encoding.UTF8.GetBytes("test val 43") }, dh4);
+
+                producer.Flush(TimeSpan.FromSeconds(10));
+            }
+
+            Assert.Equal(2, count);
 
             Assert.Equal(0, Library.HandleCount);
             LogToFile("end   Producer_Produce");

diff --git a/test/Confluent.Kafka.IntegrationTests/Tests/Producer_ProduceAsync_Error.cs b/test/Confluent.Kafka.IntegrationTests/Tests/Producer_ProduceAsync_Error.cs
@@ -106,6 +106,40 @@ public void Producer_ProduceAsync_Error(string bootstrapServers)
                 Assert.Equal(TimestampType.NotAvailable, dr.Message.Timestamp.Type);
             }
 
+            // byte[] case
+
+            Task<DeliveryResult<ReadOnlyMemory<byte>, Memory<byte>>> drt3;
+            using (var producer = new TestProducerBuilder<ReadOnlyMemory<byte>, Memory<byte>>(producerConfig).Build())
+            {
+                drt3 = producer.ProduceAsync(
+                    new TopicPartition(partitionedTopic, 42),
+                    new Message<ReadOnlyMemory<byte>, Memory<byte>> { Key = new byte[] { 100 }, Value = new byte[] { 101 } });
+                Assert.Equal(0, producer.Flush(TimeSpan.FromSeconds(10)));
+            }
+
+            Assert.Throws<AggregateException>(() => { drt.Wait(); });
+
+            try
+            {
+                _ = drt3.Result;
+            }
+            catch (AggregateException e)
+            {
+                var inner = e.InnerException;
+                Assert.IsType<ProduceException<ReadOnlyMemory<byte>, Memory<byte>>>(inner);
+                var dr = ((ProduceException<ReadOnlyMemory<byte>, Memory<byte>>)inner).DeliveryResult;
+                var err = ((ProduceException<ReadOnlyMemory<byte>, Memory<byte>>)inner).Error;
+
+                Assert.True(err.IsError);
+                Assert.False(err.IsFatal);
+                Assert.Equal(partitionedTopic, dr.Topic);
+                Assert.Equal(Offset.Unset, dr.Offset);
+                Assert.True(dr.Partition == 42);
+                Assert.Equal(new byte[] { 100 }, dr.Message.Key.ToArray());
+                Assert.Equal(new byte[] { 101 }, dr.Message.Value.ToArray());
+                Assert.Equal(TimestampType.NotAvailable, dr.Message.Timestamp.Type);
+            }
+
             Assert.Equal(0, Library.HandleCount);
             LogToFile("end   Producer_ProduceAsync_Error");
         }

diff --git a/test/Confluent.Kafka.IntegrationTests/Tests/Producer_ProduceAsync_Null_Task.cs b/test/Confluent.Kafka.IntegrationTests/Tests/Producer_ProduceAsync_Null_Task.cs
@@ -90,6 +90,29 @@ public void Producer_ProduceAsync_Null_Task(string bootstrapServers)
 
             Assert.Equal((Partition)1, drs2[0].Result.Partition);
 
+            // Memory<byte>? case
+
+            var drs3 = new List<Task<DeliveryResult<Memory<byte>?, Memory<byte>?>>>();
+            using (var producer = new TestProducerBuilder<Memory<byte>?, Memory<byte>?>(producerConfig).Build())
+            {
+                drs3.Add(producer.ProduceAsync(new TopicPartition(partitionedTopic, 1), new Message<Memory<byte>?, Memory<byte>?>()));
+                drs3.Add(producer.ProduceAsync(partitionedTopic, new Message<Memory<byte>?, Memory<byte>?>()));
+                Assert.Equal(0, producer.Flush(TimeSpan.FromSeconds(10)));
+            }
+
+            for (int i = 0; i < 2; ++i)
+            {
+                var dr = drs3[i].Result;
+                Assert.True(dr.Partition == 0 || dr.Partition == 1);
+                Assert.Equal(partitionedTopic, dr.Topic);
+                Assert.True(dr.Offset >= 0);
+                Assert.Null(dr.Message.Key);
+                Assert.Null(dr.Message.Value);
+                Assert.Equal(TimestampType.CreateTime, dr.Message.Timestamp.Type);
+                Assert.True(Math.Abs((DateTime.UtcNow - dr.Message.Timestamp.UtcDateTime).TotalMinutes) < 1.0);
+            }
+
+            Assert.Equal((Partition)1, drs3[0].Result.Partition);
 
             Assert.Equal(0, Library.HandleCount);
             LogToFile("end   Producer_ProduceAsync_Null_Task");

diff --git a/test/Confluent.Kafka.IntegrationTests/Tests/Producer_ProduceAsync_Task.cs b/test/Confluent.Kafka.IntegrationTests/Tests/Producer_ProduceAsync_Task.cs
@@ -98,6 +98,64 @@ public void Producer_ProduceAsync_Task(string bootstrapServers)
 
             Assert.Equal((Partition)1, drs2[0].Result.Partition);
 
+            // Memory<byte> case
+
+            var drs3 = new List<Task<DeliveryResult<Memory<byte>, ReadOnlyMemory<byte>>>>();
+            using (var producer = new TestProducerBuilder<Memory<byte>, ReadOnlyMemory<byte>>(producerConfig).Build())
+            {
+                drs3.Add(producer.ProduceAsync(
+                    new TopicPartition(partitionedTopic, 1),
+                    new Message<Memory<byte>, ReadOnlyMemory<byte>> { Key = Encoding.UTF8.GetBytes("test key 2"), Value = Encoding.UTF8.GetBytes("test val 2") }));
+                drs3.Add(producer.ProduceAsync(
+                    partitionedTopic,
+                    new Message<Memory<byte>, ReadOnlyMemory<byte>> { Key = Encoding.UTF8.GetBytes("test key 3"), Value = Encoding.UTF8.GetBytes("test val 3") }));
+                Assert.Equal(0, producer.Flush(TimeSpan.FromSeconds(10)));
+            }
+
+            for (int i = 0; i < 2; ++i)
+            {
+                var dr = drs3[i].Result;
+                Assert.Equal(partitionedTopic, dr.Topic);
+                Assert.True(dr.Offset >= 0);
+                Assert.True(dr.Partition == 0 || dr.Partition == 1);
+                Assert.Equal($"test key {i+2}", Encoding.UTF8.GetString(dr.Message.Key.Span));
+                Assert.Equal($"test val {i+2}", Encoding.UTF8.GetString(dr.Message.Value.Span));
+                Assert.Equal(TimestampType.CreateTime, dr.Message.Timestamp.Type);
+                Assert.True(Math.Abs((DateTime.UtcNow - dr.Message.Timestamp.UtcDateTime).TotalMinutes) < 1.0);
+            }
+
+            Assert.Equal((Partition)1, drs3[0].Result.Partition);
+
+            // Memory<byte>? case
+
+            var drs4 = new List<Task<DeliveryResult<ReadOnlyMemory<byte>?, Memory<byte>?>>>();
+            using (var producer = new TestProducerBuilder<ReadOnlyMemory<byte>?, Memory<byte>?>(producerConfig).Build())
+            {
+                drs4.Add(producer.ProduceAsync(
+                    new TopicPartition(partitionedTopic, 1),
+                    new Message<ReadOnlyMemory<byte>?, Memory<byte>?> { Key = Encoding.UTF8.GetBytes("test key 2"), Value = Encoding.UTF8.GetBytes("test val 2") }));
+                drs4.Add(producer.ProduceAsync(
+                    partitionedTopic,
+                    new Message<ReadOnlyMemory<byte>?, Memory<byte>?> { Key = Encoding.UTF8.GetBytes("test key 3"), Value = Encoding.UTF8.GetBytes("test val 3") }));
+                Assert.Equal(0, producer.Flush(TimeSpan.FromSeconds(10)));
+            }
+
+            for (int i = 0; i < 2; ++i)
+            {
+                var dr = drs4[i].Result;
+                Assert.Equal(partitionedTopic, dr.Topic);
+                Assert.True(dr.Offset >= 0);
+                Assert.True(dr.Partition == 0 || dr.Partition == 1);
+                Assert.True(dr.Message.Key.HasValue);
+                Assert.Equal($"test key {i+2}", Encoding.UTF8.GetString(dr.Message.Key.Value.Span));
+                Assert.True(dr.Message.Value.HasValue);
+                Assert.Equal($"test val {i+2}", Encoding.UTF8.GetString(dr.Message.Value.Value.Span));
+                Assert.Equal(TimestampType.CreateTime, dr.Message.Timestamp.Type);
+                Assert.True(Math.Abs((DateTime.UtcNow - dr.Message.Timestamp.UtcDateTime).TotalMinutes) < 1.0);
+            }
+
+            Assert.Equal((Partition)1, drs4[0].Result.Partition);
+
             Assert.Equal(0, Library.HandleCount);
             LogToFile("end   Producer_ProduceAsync_Task");
         }

diff --git a/test/Confluent.Kafka.IntegrationTests/Tests/Producer_Produce_Async.cs b/test/Confluent.Kafka.IntegrationTests/Tests/Producer_Produce_Async.cs
@@ -60,36 +60,5 @@ public void Producer_Produce_Async(string bootstrapServers)
             Assert.Equal(0, Library.HandleCount);
             LogToFile("end   Producer_Produce_Async");
         }
-
-        [Theory, MemberData(nameof(KafkaParameters))]
-        public void Producer_Produce_Memory_Async(string bootstrapServers)
-        {
-            LogToFile("start Producer_Produce_Memory_Async");
-
-            var producerConfig = new ProducerConfig { BootstrapServers = bootstrapServers };
-
-            using (var testTopic = new TemporaryTopic(bootstrapServers, 1))
-            using (var producer = new TestProducerBuilder<Memory<byte>?, Memory<byte>>(producerConfig)
-                .Build())
-            using (var dProducer = new DependentProducerBuilder<ReadOnlyMemory<byte>, ReadOnlyMemory<byte>?>(producer.Handle)
-                .Build())
-            {
-                Memory<byte> data = new byte[] { 1, 2, 3, 4 };
-                Assert.Throws<ProduceException<Memory<byte>?, Memory<byte>>>(
-                    () => producer.Produce(testTopic.Name, new Message<Memory<byte>?, Memory<byte>> { Value = data }));
-
-                Assert.Throws<ProduceException<Memory<byte>?, Memory<byte>>>(
-                    () => producer.Produce(testTopic.Name, new Message<Memory<byte>?, Memory<byte>> { Value = data }, dr => { Assert.True(false); }));
-
-                Assert.Throws<ProduceException<ReadOnlyMemory<byte>, ReadOnlyMemory<byte>?>>(
-                    () => dProducer.Produce(testTopic.Name, new Message<ReadOnlyMemory<byte>, ReadOnlyMemory<byte>?> { Key = data }));
-
-                Assert.Throws<ProduceException<ReadOnlyMemory<byte>, ReadOnlyMemory<byte>?>>(
-                    () => dProducer.Produce(testTopic.Name, new Message<ReadOnlyMemory<byte>, ReadOnlyMemory<byte>?> { Key = data }, dr => { Assert.True(false); }));
-            }
-
-            Assert.Equal(0, Library.HandleCount);
-            LogToFile("end   Producer_Produce_Memory_Async");
-        }
     }
 }
diff --git a/test/Confluent.Kafka.IntegrationTests/Tests/Producer_Produce_Error.cs b/test/Confluent.Kafka.IntegrationTests/Tests/Producer_Produce_Error.cs
@@ -89,6 +89,29 @@ public void Producer_Produce_Error(string bootstrapServers)
 
             Assert.Equal(1, count);
 
+            // Memory<byte> case.
+
+            count = 0;
+            Action<DeliveryReport<Memory<byte>, ReadOnlyMemory<byte>?>> dh3 = dr =>
+            {
+                Assert.Equal(ErrorCode.Local_UnknownPartition, dr.Error.Code);
+                Assert.Equal((Partition)42, dr.Partition);
+                Assert.Equal(singlePartitionTopic, dr.Topic);
+                Assert.Equal(Offset.Unset, dr.Offset);
+                Assert.Equal(new byte[] { 11 }, dr.Message.Key.ToArray());
+                Assert.Null(dr.Message.Value);
+                Assert.Equal(TimestampType.NotAvailable, dr.Message.Timestamp.Type);
+                count += 1;
+            };
+
+            using (var producer = new TestProducerBuilder<Memory<byte>, ReadOnlyMemory<byte>?>(producerConfig).Build())
+            {
+                producer.Produce(new TopicPartition(singlePartitionTopic, 42), new Message<Memory<byte>, ReadOnlyMemory<byte>?> { Key = new byte[] { 11 } }, dh3);
+                producer.Flush(TimeSpan.FromSeconds(10));
+            }
+
+            Assert.Equal(1, count);
+
             Assert.Equal(0, Library.HandleCount);
             LogToFile("end   Producer_Produce_Error");
         }

diff --git a/test/Confluent.Kafka.IntegrationTests/Tests/Producer_Produce_Null.cs b/test/Confluent.Kafka.IntegrationTests/Tests/Producer_Produce_Null.cs
@@ -89,6 +89,31 @@ public void Producer_Produce_Null(string bootstrapServers)
 
             Assert.Equal(2, count);
 
+            // Memory<byte>? case.
+
+            count = 0;
+            Action<DeliveryReport<ReadOnlyMemory<byte>?, Memory<byte>?>> dh3 = dr =>
+            {
+                Assert.Equal(ErrorCode.NoError, dr.Error.Code);
+                Assert.Equal((Partition)0, dr.Partition);
+                Assert.Equal(singlePartitionTopic, dr.Topic);
+                Assert.True(dr.Offset >= 0);
+                Assert.Null(dr.Message.Key);
+                Assert.Null(dr.Message.Value);
+                Assert.Equal(TimestampType.CreateTime, dr.Message.Timestamp.Type);
+                Assert.True(Math.Abs((DateTime.UtcNow - dr.Message.Timestamp.UtcDateTime).TotalMinutes) < 1.0);
+                count += 1;
+            };
+
+            using (var producer = new TestProducerBuilder<ReadOnlyMemory<byte>?, Memory<byte>?>(producerConfig).Build())
+            {
+                producer.Produce(new TopicPartition(singlePartitionTopic, 0), new Message<ReadOnlyMemory<byte>?, Memory<byte>?>(), dh3);
+                producer.Produce(singlePartitionTopic, new Message<ReadOnlyMemory<byte>?, Memory<byte>?>(), dh3);
+                producer.Flush(TimeSpan.FromSeconds(10));
+            }
+
+            Assert.Equal(2, count);
+
             Assert.Equal(0, Library.HandleCount);
             LogToFile("end   Producer_Produce_Null");
         }