Updates test execution to avoid failures (#258)

* Use `SingleOrDefault` instead of `Single` to manage conditions where it is raised `System.InvalidOperationException: Sequence contains no elements` * Split windows jobs since Kafka seems to crash on that OS if heavily loaded * In case of timeout, especially in Windows, exit without report error because the error comes from external source * Configuration in workflow folder and remove prefetch * Upload dumps * Added KafkaStreams test * Update management of exceptions * Added Raw and Buffered tests for KafkaStreams * Use latest JDK for tests * Code harmonization * Adds tests with prefetch * Fix #22 aligning key of headers to the JVM type * Update tests moving out execution from Main to verify if some tests fails for the same reason reported in masesgroup/KNet#509
masesgroup · Jul 2, 2024 · 3aba45f · 3aba45f
1 parent f976afb
commit 3aba45f
Show file tree

Hide file tree

Showing 26 changed files with 361 additions and 131 deletions.
diff --git a/.github/workflows/build.yaml b/.github/workflows/build.yaml
diff --git a/.github/workflows/configuration/Benchmark.KNetReplicator.json b/.github/workflows/configuration/Benchmark.KNetReplicator.json
@@ -0,0 +1,6 @@
+{
+  "DatabaseName": "TestDBBenchmark",
+  "BootstrapServers": "localhost:9092",
+  "NumberOfExecutions": 10,
+  "UseEnumeratorWithPrefetch": false
+}
diff --git a/.github/workflows/configuration/Benchmark.KNetStreams.Buffered.Prefetch.json b/.github/workflows/configuration/Benchmark.KNetStreams.Buffered.Prefetch.json
@@ -0,0 +1,6 @@
+{
+  "DatabaseName": "TestDBBenchmark",
+  "UseCompactedReplicator": false,
+  "BootstrapServers": "localhost:9092",
+  "NumberOfExecutions": 10
+}
diff --git a/.github/workflows/configuration/Benchmark.KNetStreams.Buffered.json b/.github/workflows/configuration/Benchmark.KNetStreams.Buffered.json
@@ -0,0 +1,7 @@
+{
+  "DatabaseName": "TestDBBenchmark",
+  "UseCompactedReplicator": false,
+  "BootstrapServers": "localhost:9092",
+  "NumberOfExecutions": 10,
+  "UseEnumeratorWithPrefetch": false
+}
diff --git a/.github/workflows/configuration/Benchmark.KNetStreams.Raw.Prefetch.json b/.github/workflows/configuration/Benchmark.KNetStreams.Raw.Prefetch.json
@@ -0,0 +1,7 @@
+{
+  "DatabaseName": "TestDBBenchmark",
+  "UseCompactedReplicator": false,
+  "UseByteBufferDataTransfer": false,
+  "BootstrapServers": "localhost:9092",
+  "NumberOfExecutions": 10
+}
diff --git a/.github/workflows/configuration/Benchmark.KNetStreams.Raw.json b/.github/workflows/configuration/Benchmark.KNetStreams.Raw.json
@@ -0,0 +1,8 @@
+{
+  "DatabaseName": "TestDBBenchmark",
+  "UseCompactedReplicator": false,
+  "UseByteBufferDataTransfer": false,
+  "BootstrapServers": "localhost:9092",
+  "NumberOfExecutions": 10,
+  "UseEnumeratorWithPrefetch": false
+}
diff --git a/.github/workflows/configuration/Benchmark.KafkaStreams.Buffered.json b/.github/workflows/configuration/Benchmark.KafkaStreams.Buffered.json
@@ -0,0 +1,7 @@
+{
+  "DatabaseName": "TestDBBenchmark",
+  "UseCompactedReplicator": false,
+  "UseKNetStreams": false,
+  "BootstrapServers": "localhost:9092",
+  "NumberOfExecutions": 10
+}
diff --git a/.github/workflows/configuration/Benchmark.KafkaStreams.Raw.json b/.github/workflows/configuration/Benchmark.KafkaStreams.Raw.json
@@ -0,0 +1,8 @@
+{
+  "DatabaseName": "TestDBBenchmark",
+  "UseCompactedReplicator": false,
+  "UseKNetStreams": false,
+  "UseByteBufferDataTransfer": false,
+  "BootstrapServers": "localhost:9092",
+  "NumberOfExecutions": 10
+}
diff --git a/.github/workflows/log4j.properties → .../workflows/configuration/log4j.properties b/.github/workflows/log4j.properties → .../workflows/configuration/log4j.properties
diff --git a/.github/workflows/server.properties → ...workflows/configuration/server.properties b/.github/workflows/server.properties → ...workflows/configuration/server.properties
diff --git a/.github/workflows/zookeeper.properties → ...kflows/configuration/zookeeper.properties b/.github/workflows/zookeeper.properties → ...kflows/configuration/zookeeper.properties
diff --git a/src/net/KEFCore.SerDes.Avro/AvroKEFCoreSerDes.cs b/src/net/KEFCore.SerDes.Avro/AvroKEFCoreSerDes.cs
@@ -127,8 +127,8 @@ public override byte[] Serialize(string topic, TData data)
                 /// <inheritdoc cref="SerDes{TData, TJVM}.SerializeWithHeaders(string, Headers, TData)"/>
                 public override byte[] SerializeWithHeaders(string topic, Headers headers, TData data)
                 {
-                    headers?.Add(KNetSerialization.KeyTypeIdentifier, keyTypeName);
-                    headers?.Add(KNetSerialization.KeySerializerIdentifier, keySerDesName);
+                    headers?.Add(KNetSerialization.KeyTypeIdentifierJVM, keyTypeName);
+                    headers?.Add(KNetSerialization.KeySerializerIdentifierJVM, keySerDesName);
 
                     if (_defaultSerDes != null) return _defaultSerDes.SerializeWithHeaders(topic, headers, data);
 
@@ -199,8 +199,8 @@ public override ByteBuffer Serialize(string topic, TData data)
                 /// <inheritdoc cref="SerDes{TData, TJVM}.SerializeWithHeaders(string, Headers, TData)"/>
                 public override ByteBuffer SerializeWithHeaders(string topic, Headers headers, TData data)
                 {
-                    headers?.Add(KNetSerialization.KeyTypeIdentifier, keyTypeName);
-                    headers?.Add(KNetSerialization.KeySerializerIdentifier, keySerDesName);
+                    headers?.Add(KNetSerialization.KeyTypeIdentifierJVM, keyTypeName);
+                    headers?.Add(KNetSerialization.KeySerializerIdentifierJVM, keySerDesName);
 
                     if (_defaultSerDes != null) return _defaultSerDes.SerializeWithHeaders(topic, headers, data);
 
@@ -310,8 +310,8 @@ public override byte[] Serialize(string topic, TData data)
                 /// <inheritdoc cref="SerDes{TData, TJVM}.SerializeWithHeaders(string, Headers, TData)"/>
                 public override byte[] SerializeWithHeaders(string topic, Headers headers, TData data)
                 {
-                    headers?.Add(KNetSerialization.KeyTypeIdentifier, keyTypeName);
-                    headers?.Add(KNetSerialization.KeySerializerIdentifier, keySerDesName);
+                    headers?.Add(KNetSerialization.KeyTypeIdentifierJVM, keyTypeName);
+                    headers?.Add(KNetSerialization.KeySerializerIdentifierJVM, keySerDesName);
 
                     if (_defaultSerDes != null) return _defaultSerDes.SerializeWithHeaders(topic, headers, data);
 
@@ -377,8 +377,8 @@ public override ByteBuffer Serialize(string topic, TData data)
                 /// <inheritdoc cref="SerDes{TData, TJVM}.SerializeWithHeaders(string, Headers, TData)"/>
                 public override ByteBuffer SerializeWithHeaders(string topic, Headers headers, TData data)
                 {
-                    headers?.Add(KNetSerialization.KeyTypeIdentifier, keyTypeName);
-                    headers?.Add(KNetSerialization.KeySerializerIdentifier, keySerDesName);
+                    headers?.Add(KNetSerialization.KeyTypeIdentifierJVM, keyTypeName);
+                    headers?.Add(KNetSerialization.KeySerializerIdentifierJVM, keySerDesName);
 
                     if (_defaultSerDes != null) return _defaultSerDes.SerializeWithHeaders(topic, headers, data);
 
@@ -494,8 +494,8 @@ public override byte[] Serialize(string topic, TData data)
                 /// <inheritdoc cref="SerDes{TData, TJVM}.SerializeWithHeaders(string, Headers, TData)"/>
                 public override byte[] SerializeWithHeaders(string topic, Headers headers, TData data)
                 {
-                    headers?.Add(KNetSerialization.ValueSerializerIdentifier, valueContainerSerDesName);
-                    headers?.Add(KNetSerialization.ValueTypeIdentifier, valueContainerName);
+                    headers?.Add(KNetSerialization.ValueSerializerIdentifierJVM, valueContainerSerDesName);
+                    headers?.Add(KNetSerialization.ValueTypeIdentifierJVM, valueContainerName);
 
                     using MemoryStream memStream = new();
                     BinaryEncoder encoder = new(memStream);
@@ -562,8 +562,8 @@ public override ByteBuffer Serialize(string topic, TData data)
                 /// <inheritdoc cref="SerDes{TData, TJVM}.SerializeWithHeaders(string, Headers, TData)"/>
                 public override ByteBuffer SerializeWithHeaders(string topic, Headers headers, TData data)
                 {
-                    headers?.Add(KNetSerialization.ValueSerializerIdentifier, valueContainerSerDesName);
-                    headers?.Add(KNetSerialization.ValueTypeIdentifier, valueContainerName);
+                    headers?.Add(KNetSerialization.ValueSerializerIdentifierJVM, valueContainerSerDesName);
+                    headers?.Add(KNetSerialization.ValueTypeIdentifierJVM, valueContainerName);
 
                     MemoryStream memStream = new();
                     BinaryEncoder encoder = new(memStream);
@@ -668,8 +668,8 @@ public override byte[] Serialize(string topic, TData data)
                 /// <inheritdoc cref="SerDes{TData, TJVM}.SerializeWithHeaders(string, Headers, TData)"/>
                 public override byte[] SerializeWithHeaders(string topic, Headers headers, TData data)
                 {
-                    headers?.Add(KNetSerialization.ValueSerializerIdentifier, valueContainerSerDesName);
-                    headers?.Add(KNetSerialization.ValueTypeIdentifier, valueContainerName);
+                    headers?.Add(KNetSerialization.ValueSerializerIdentifierJVM, valueContainerSerDesName);
+                    headers?.Add(KNetSerialization.ValueTypeIdentifierJVM, valueContainerName);
 
                     using MemoryStream memStream = new();
                     JsonEncoder encoder = new(AvroValueContainer._SCHEMA, memStream);
@@ -737,8 +737,8 @@ public override ByteBuffer Serialize(string topic, TData data)
                 /// <inheritdoc cref="SerDes{TData, TJVM}.SerializeWithHeaders(string, Headers, TData)"/>
                 public override ByteBuffer SerializeWithHeaders(string topic, Headers headers, TData data)
                 {
-                    headers?.Add(KNetSerialization.ValueSerializerIdentifier, valueContainerSerDesName);
-                    headers?.Add(KNetSerialization.ValueTypeIdentifier, valueContainerName);
+                    headers?.Add(KNetSerialization.ValueSerializerIdentifierJVM, valueContainerSerDesName);
+                    headers?.Add(KNetSerialization.ValueTypeIdentifierJVM, valueContainerName);
 
                     MemoryStream memStream = new();
                     JsonEncoder encoder = new(AvroValueContainer._SCHEMA, memStream);

diff --git a/src/net/KEFCore.SerDes.Protobuf/ProtobufKEFCoreSerDes.cs b/src/net/KEFCore.SerDes.Protobuf/ProtobufKEFCoreSerDes.cs
@@ -118,8 +118,8 @@ public override byte[] Serialize(string topic, TData data)
             /// <inheritdoc cref="SerDes{TData, TJVM}.SerializeWithHeaders(string, Headers, TData)"/>
             public override byte[] SerializeWithHeaders(string topic, Headers headers, TData data)
             {
-                headers?.Add(KNetSerialization.KeyTypeIdentifier, keyTypeName);
-                headers?.Add(KNetSerialization.KeySerializerIdentifier, keySerDesName);
+                headers?.Add(KNetSerialization.KeyTypeIdentifierJVM, keyTypeName);
+                headers?.Add(KNetSerialization.KeySerializerIdentifierJVM, keySerDesName);
 
                 if (_defaultSerDes != null) return _defaultSerDes.SerializeWithHeaders(topic, headers, data);
                 KeyContainer keyContainer = null!;
@@ -185,8 +185,8 @@ public override ByteBuffer Serialize(string topic, TData data)
             /// <inheritdoc cref="SerDes{TData, TJVM}.SerializeWithHeaders(string, Headers, TData)"/>
             public override ByteBuffer SerializeWithHeaders(string topic, Headers headers, TData data)
             {
-                headers?.Add(KNetSerialization.KeyTypeIdentifier, keyTypeName);
-                headers?.Add(KNetSerialization.KeySerializerIdentifier, keySerDesName);
+                headers?.Add(KNetSerialization.KeyTypeIdentifierJVM, keyTypeName);
+                headers?.Add(KNetSerialization.KeySerializerIdentifierJVM, keySerDesName);
 
                 if (_defaultSerDes != null) return _defaultSerDes.SerializeWithHeaders(topic, headers, data);
                 KeyContainer keyContainer = null!;
@@ -298,8 +298,8 @@ public override byte[] Serialize(string topic, TData data)
             /// <inheritdoc cref="SerDes{TData, TJVM}.SerializeWithHeaders(string, Headers, TData)"/>
             public override byte[] SerializeWithHeaders(string topic, Headers headers, TData data)
             {
-                headers?.Add(KNetSerialization.ValueSerializerIdentifier, valueContainerSerDesName);
-                headers?.Add(KNetSerialization.ValueTypeIdentifier, valueContainerName);
+                headers?.Add(KNetSerialization.ValueSerializerIdentifierJVM, valueContainerSerDesName);
+                headers?.Add(KNetSerialization.ValueTypeIdentifierJVM, valueContainerName);
 
                 using (MemoryStream stream = new())
                 {
@@ -361,8 +361,8 @@ public override ByteBuffer Serialize(string topic, TData data)
             /// <inheritdoc cref="SerDes{TData, TJVM}.SerializeWithHeaders(string, Headers, TData)"/>
             public override ByteBuffer SerializeWithHeaders(string topic, Headers headers, TData data)
             {
-                headers?.Add(KNetSerialization.ValueSerializerIdentifier, valueContainerSerDesName);
-                headers?.Add(KNetSerialization.ValueTypeIdentifier, valueContainerName);
+                headers?.Add(KNetSerialization.ValueSerializerIdentifierJVM, valueContainerSerDesName);
+                headers?.Add(KNetSerialization.ValueTypeIdentifierJVM, valueContainerName);
 
                 MemoryStream stream = new();
                 data.WriteTo(stream);

diff --git a/src/net/KEFCore.SerDes/DefaultKEFCoreSerDes.cs b/src/net/KEFCore.SerDes/DefaultKEFCoreSerDes.cs
@@ -126,8 +126,8 @@ public override byte[] Serialize(string topic, TData data)
             /// <inheritdoc cref="SerDes{TData, TJVM}.SerializeWithHeaders(string, Headers, TData)"/>
             public override byte[] SerializeWithHeaders(string topic, Headers headers, TData data)
             {
-                headers?.Add(KNetSerialization.KeyTypeIdentifier, keyTypeName);
-                headers?.Add(KNetSerialization.KeySerializerIdentifier, keySerDesName);
+                headers?.Add(KNetSerialization.KeyTypeIdentifierJVM, keyTypeName);
+                headers?.Add(KNetSerialization.KeySerializerIdentifierJVM, keySerDesName);
 
                 if (_defaultSerDes != null) return _defaultSerDes.SerializeWithHeaders(topic, headers, data);
                 var jsonStr = System.Text.Json.JsonSerializer.Serialize<TData>(data);
@@ -191,8 +191,8 @@ public override ByteBuffer Serialize(string topic, TData data)
             /// <inheritdoc cref="SerDes{TData, TJVM}.SerializeWithHeaders(string, Headers, TData)"/>
             public override ByteBuffer SerializeWithHeaders(string topic, Headers headers, TData data)
             {
-                headers?.Add(KNetSerialization.KeyTypeIdentifier, keyTypeName);
-                headers?.Add(KNetSerialization.KeySerializerIdentifier, keySerDesName);
+                headers?.Add(KNetSerialization.KeyTypeIdentifierJVM, keyTypeName);
+                headers?.Add(KNetSerialization.KeySerializerIdentifierJVM, keySerDesName);
 
                 if (_defaultSerDes != null) return _defaultSerDes.SerializeWithHeaders(topic, headers, data);
 
@@ -301,8 +301,8 @@ public override byte[] Serialize(string topic, TData data)
             /// <inheritdoc cref="SerDes{TData, TJVM}.SerializeWithHeaders(string, Headers, TData)"/>
             public override byte[] SerializeWithHeaders(string topic, Headers headers, TData data)
             {
-                headers?.Add(KNetSerialization.ValueSerializerIdentifier, valueContainerSerDesName);
-                headers?.Add(KNetSerialization.ValueTypeIdentifier, valueContainerName);
+                headers?.Add(KNetSerialization.ValueSerializerIdentifierJVM, valueContainerSerDesName);
+                headers?.Add(KNetSerialization.ValueTypeIdentifierJVM, valueContainerName);
 
                 var jsonStr = System.Text.Json.JsonSerializer.Serialize<TData>(data, _options);
                 return Encoding.UTF8.GetBytes(jsonStr);
@@ -365,8 +365,8 @@ public override ByteBuffer Serialize(string topic, TData data)
             /// <inheritdoc cref="SerDes{TData, TJVM}.SerializeWithHeaders(string, Headers, TData)"/>
             public override ByteBuffer SerializeWithHeaders(string topic, Headers headers, TData data)
             {
-                headers?.Add(KNetSerialization.ValueSerializerIdentifier, valueContainerSerDesName);
-                headers?.Add(KNetSerialization.ValueTypeIdentifier, valueContainerName);
+                headers?.Add(KNetSerialization.ValueSerializerIdentifierJVM, valueContainerSerDesName);
+                headers?.Add(KNetSerialization.ValueTypeIdentifierJVM, valueContainerName);
 
                 var ms = new MemoryStream();
                 System.Text.Json.JsonSerializer.Serialize<TData>(ms, data, _options);

diff --git a/src/net/KEFCore.SerDes/KEFCoreSerDes.Helpers.cs b/src/net/KEFCore.SerDes/KEFCoreSerDes.Helpers.cs
@@ -151,22 +151,22 @@ public static object FromRecord(Org.Apache.Kafka.Clients.Consumer.ConsumerRecord
             foreach (var header in headers.ToArray())
             {
                 var key = header.Key();
-                if (key == KNetSerialization.KeyTypeIdentifier)
+                if (key == KNetSerialization.KeyTypeIdentifierJVM)
                 {
                     var strType = Encoding.UTF8.GetString(header.Value());
                     keyType = Type.GetType(strType, true)!;
                 }
-                if (key == KNetSerialization.KeySerializerIdentifier)
+                if (key == KNetSerialization.KeySerializerIdentifierJVM)
                 {
                     var strType = Encoding.UTF8.GetString(header.Value());
                     keySerializerSelectorType = Type.GetType(strType, true)!;
                 }
-                if (key == KNetSerialization.ValueTypeIdentifier)
+                if (key == KNetSerialization.ValueTypeIdentifierJVM)
                 {
                     var strType = Encoding.UTF8.GetString(header.Value());
                     valueType = Type.GetType(strType, true)!;
                 }
-                if (key == KNetSerialization.ValueSerializerIdentifier)
+                if (key == KNetSerialization.ValueSerializerIdentifierJVM)
                 {
                     var strType = Encoding.UTF8.GetString(header.Value());
                     valueSerializerSelectorType = Type.GetType(strType, true)!;

diff --git a/src/net/KEFCore/Storage/Internal/KNetStreamsRetriever.cs b/src/net/KEFCore/Storage/Internal/KNetStreamsRetriever.cs
@@ -444,20 +444,20 @@ public ValueTask<bool> MoveNextAsync()
                 {
                     _moveNextSw.Start();
 #endif
-            ValueTask<bool> hasNext = _asyncEnumerator.MoveNextAsync();
+            ValueTask<bool> hasNext = _asyncEnumerator == null ? new ValueTask<bool>(false) : _asyncEnumerator.MoveNextAsync();
             hasNext.AsTask().Wait();
             if (hasNext.Result)
             {
 #if DEBUG_PERFORMANCE
                 _cycles++;
                 _valueGetSw.Start();
 #endif
-                KeyValue<TKey, TValue, TJVMKey, TJVMValue> kv =  _asyncEnumerator.Current;
+                KeyValue<TKey, TValue, TJVMKey, TJVMValue>? kv =  _asyncEnumerator?.Current;
 #if DEBUG_PERFORMANCE
                 _valueGetSw.Stop();
                 _valueGet2Sw.Start();
 #endif
-                TValue value = kv.Value;
+                TValue value = kv != null ? kv.Value : default;
 #if DEBUG_PERFORMANCE
                 _valueGet2Sw.Stop();
                 _valueBufferSw.Start();

diff --git a/src/net/KEFCore/Storage/Internal/KafkaStreamsBaseRetriever.cs b/src/net/KEFCore/Storage/Internal/KafkaStreamsBaseRetriever.cs
@@ -90,7 +90,7 @@ public KafkaStreamsBaseRetriever(IKafkaCluster kafkaCluster, IEntityType entityT
         _builder ??= builder;
         _topicName = _entityType.TopicName(kafkaCluster.Options);
         _usePersistentStorage = _kafkaCluster.Options.UsePersistentStorage;
-        _properties ??= _kafkaCluster.Options.StreamsOptions(_kafkaCluster.Options.ApplicationId);
+        _properties ??= _kafkaCluster.Options.StreamsOptions(_entityType);
 
         string storageId = _entityType.StorageIdForTable(_kafkaCluster.Options);
         _storageId = _usePersistentStorage ? storageId : Process.GetCurrentProcess().ProcessName + "-" + storageId;