Merge branch 'develop' into feature/processor-api

yasinparsi · Oct 18, 2022 · 646f61b · 646f61b
2 parents 4870e88 + d888046
commit 646f61b
Show file tree

Hide file tree

Showing 82 changed files with 1,222 additions and 505 deletions.
diff --git a/.circleci/config.yml b/.circleci/config.yml
@@ -23,7 +23,7 @@
         # END Dependencies for RocksDB
         - run: set JAVA_HOME /usr/lib/jvm/java-11-openjdk-amd64/
         - run: export JAVA_HOME
-        - run: dotnet sonarscanner begin /k:LGouellec_kafka-streams-dotnet /o:kafka-streams-dotnet /d:sonar.login=${SONAR_TOKEN} /d:sonar.host.url=https://sonarcloud.io /d:sonar.cs.opencover.reportsPaths="**\coverage*.opencover.xml" /d:sonar.coverage.exclusions="**sample*/*.cs,**test*/*.cs,**Tests*.cs"
+        - run: dotnet sonarscanner begin /k:LGouellec_kafka-streams-dotnet /o:kafka-streams-dotnet /d:sonar.login=${SONAR_TOKEN} /d:sonar.host.url=https://sonarcloud.io /d:sonar.cs.opencover.reportsPaths="**\coverage*.opencover.xml" /d:sonar.coverage.exclusions="**sample*/*.cs,**test*/*.cs,**Tests*.cs,**Mock*.cs"
         - run: dotnet build
         - run: dotnet test --no-restore --no-build --verbosity normal -f net6.0 --collect:"XPlat Code Coverage" /p:CollectCoverage=true /p:CoverletOutputFormat=opencover test/Streamiz.Kafka.Net.Tests/Streamiz.Kafka.Net.Tests.csproj
         - run: dotnet sonarscanner end /d:sonar.login=${SONAR_TOKEN}

diff --git a/.github/workflows/integration.yml b/.github/workflows/integration.yml
@@ -10,6 +10,7 @@ jobs:
     runs-on: ubuntu-latest
 
     steps:
+    - uses: actions/checkout@v2
     - name: Setup .NET 6.0
       uses: actions/setup-dotnet@v1
       with:

diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml
@@ -41,4 +41,4 @@ jobs:
     - name: Pack Metrics Prometheus
       run: dotnet pack metrics/Streamiz.Kafka.Net.Metrics.Prometheus/Streamiz.Kafka.Net.Metrics.Prometheus.csproj --configuration Release --no-build --no-restore
     - name: Publish in nuget.org
-      run: dotnet nuget push **/*symbols.nupkg -k ${{ secrets.NUGET_PACKAGE_TOKEN }} -s https://api.nuget.org/v3/index.json -n -d   
+      run: dotnet nuget push **/*.nupkg -k ${{ secrets.NUGET_PACKAGE_TOKEN }} -s https://api.nuget.org/v3/index.json -n -d   
diff --git a/core/Crosscutting/DictionaryExtensions.cs b/core/Crosscutting/DictionaryExtensions.cs
@@ -118,6 +118,15 @@ public static bool TryAddOrUpdate<K, V>(this ConcurrentDictionary<K, V> source,
                 return source.TryAdd(key, value);
             }
         }
+
+
+        public static void CreateListOrAdd<K, V>(this IDictionary<K, List<V>> source, K key, V value)
+        {
+            if(source.ContainsKey(key))
+                source[key].Add(value);
+            else
+                source.Add(key, new List<V>{value});
+        }
 
     }
 }
diff --git a/core/KafkaStream.cs b/core/KafkaStream.cs
@@ -275,8 +275,7 @@ public override string ToString()
         /// <param name="configuration">configuration about this stream</param>
         public KafkaStream(Topology topology, IStreamConfig configuration)
             : this(topology, configuration, new DefaultKafkaClientSupplier(new KafkaLoggerAdapter(configuration), configuration))
-        {
-        }
+        { }
 
         /// <summary>
         /// Create a <see cref="KafkaStream"/> instance with your own <see cref="IKafkaSupplier" />

diff --git a/core/Mock/Kafka/MockCluster.cs b/core/Mock/Kafka/MockCluster.cs
@@ -406,7 +406,7 @@ private void NeedRebalance2(string groupId, string consumerName)
                                         if (!partitionsToRevoked.ContainsKey(consumer))
                                             partitionsToRevoked.Add(consumer, new List<TopicPartitionOffset>());
 
-                                        var tpo = consumer.TopicPartitionsOffset.FirstOrDefault(t =>
+                                        var tpo = consumer.TopicPartitionsOffset.First(t =>
                                             t.TopicPartition.Equals(part));
                                         partitionsToRevoked[consumer]
                                             .Add(new TopicPartitionOffset(part, tpo.OffsetConsumed));
@@ -619,7 +619,7 @@ private void AssignPartitionOffset(MockConsumerInformation consumerInformation,
                 if (consumerInformation.TopicPartitionsOffset.Contains(new MockTopicPartitionOffset()
                     {Partition = tp.Partition.Value, Topic = tp.Topic}))
                 {
-                    var tpo = consumerInformation.TopicPartitionsOffset.FirstOrDefault(t =>
+                    var tpo = consumerInformation.TopicPartitionsOffset.First(t =>
                         t.Partition.Equals(tp.Partition.Value) && t.Topic.Equals(tp.Topic));
                     tpo.OffsetComitted = offset;
                     tpo.OffsetConsumed = offset;
@@ -646,7 +646,7 @@ private void AssignPartition(MockConsumerInformation consumerInformation,
                 if (consumerInformation.TopicPartitionsOffset.Contains(new MockTopicPartitionOffset()
                     {Partition = tp.Partition.Value, Topic = tp.Topic}))
                 {
-                    var tpo = consumerInformation.TopicPartitionsOffset.FirstOrDefault(t =>
+                    var tpo = consumerInformation.TopicPartitionsOffset.First(t =>
                         t.Partition.Equals(tp.Partition.Value) && t.Topic.Equals(tp.Topic));
                     tpo.OffsetComitted = offset.Offset;
                     tpo.OffsetConsumed = offset.Offset;

diff --git a/core/Mock/Kafka/MockConsumer.cs b/core/Mock/Kafka/MockConsumer.cs
@@ -113,7 +113,7 @@ public Offset Position(TopicPartition partition)
         {
             var info = cluster.GetConsumerInformation(Name);
             return info != null ?
-                info.TopicPartitionsOffset.FirstOrDefault(t => t.TopicPartition.Equals(partition)).OffsetConsumed :
+                info.TopicPartitionsOffset.First(t => t.TopicPartition.Equals(partition)).OffsetConsumed :
                 Offset.Unset;
         }
 

diff --git a/core/Processors/GlobalStreamThread.cs b/core/Processors/GlobalStreamThread.cs
@@ -39,7 +39,13 @@ public StateConsumer(
             public void Initialize()
             {
                 IDictionary<TopicPartition, long> partitionOffsets = globalStateMaintainer.Initialize();
-                globalConsumer.Assign(partitionOffsets.Keys.Select(x => new TopicPartitionOffset(x, partitionOffsets[x])));
+                globalConsumer.Assign(
+                    partitionOffsets
+                        .Keys
+                        .Select(
+                            x => partitionOffsets[x] >= 0 ? 
+                                new TopicPartitionOffset(x, partitionOffsets[x] + 1 )
+                                : new TopicPartitionOffset(x, Offset.Beginning)));
 
                 lastFlush = DateTime.Now;
             }

diff --git a/core/Processors/Internal/GlobalStateManager.cs b/core/Processors/Internal/GlobalStateManager.cs
@@ -169,7 +169,7 @@ public void Register(IStateStore store, StateRestoreCallback callback)
             log.LogInformation($"Restoring state for global store {store.Name}");
             var topicPartitions = TopicPartitionsForStore(store).ToList();
             var highWatermarks = OffsetsChangelogs(topicPartitions);
-
+            
             try
             {
                 RestoreState(
@@ -181,6 +181,7 @@ public void Register(IStateStore store, StateRestoreCallback callback)
             finally
             {
                 globalConsumer.Unassign();
+                log.LogInformation($"Global store {store.Name} is completely restored");
             }
         }
 
@@ -219,7 +220,7 @@ private IEnumerable<TopicPartition> TopicPartitionsForStore(IStateStore store)
         private void RestoreState(
             StateRestoreCallback restoreCallback,
             List<TopicPartition> topicPartitions,
-            IDictionary<TopicPartition, (Offset, Offset)> highWatermarks,
+            IDictionary<TopicPartition, (Offset, Offset)> offsetWatermarks,
             Func<ConsumeResult<byte[], byte[]>, ConsumeResult<byte[], byte[]>> recordConverter)
         {
             foreach (var topicPartition in topicPartitions)
@@ -233,22 +234,26 @@ private void RestoreState(
 
                 globalConsumer.Assign((new TopicPartitionOffset(topicPartition, new Offset(checkpoint))).ToSingle());
                 offset = checkpoint;
-                highWM = highWatermarks[topicPartition].Item2;
+                var lowWM = offsetWatermarks[topicPartition].Item1;
+                highWM = offsetWatermarks[topicPartition].Item2;
 
                 while (offset < highWM - 1)
                 {
                     if (offset == Offset.Beginning && highWM == 0) // no message into local and topics;
                         break;
 
+                    if (lowWM == highWM) // if low offset == high offset
+                        break;
+
                     var records = globalConsumer.ConsumeRecords(TimeSpan.FromMilliseconds(config.PollMs),
                         config.MaxPollRestoringRecords).ToList();
 
                     var convertedRecords = records.Select(r => recordConverter(r)).ToList();
 
-                    foreach(var record in records)
+                    foreach(var record in convertedRecords)
                         restoreCallback?.Invoke(Bytes.Wrap(record.Message.Key), record.Message.Value, record.Message.Timestamp.UnixTimestampMs);
 
-                    if (records.Any())
+                    if (convertedRecords.Any())
                         offset = records.Last().Offset;
                 }
 
@@ -260,7 +265,7 @@ private void RestoreState(
         {
             return topicPartitions
                 .Select(tp => {
-                    var offsets = globalConsumer.GetWatermarkOffsets(tp);
+                    var offsets = globalConsumer.QueryWatermarkOffsets(tp, TimeSpan.FromSeconds(5));
                     return new
                     {
                         TopicPartition = tp,

diff --git a/core/Processors/Internal/InternalTopologyBuilder.cs b/core/Processors/Internal/InternalTopologyBuilder.cs
@@ -354,7 +354,7 @@ private ProcessorTopology BuildTopology(ISet<string> nodeGroup, TaskId taskId)
             // need refactor a little for repartition topic/processor source & sink etc .. change topic name
             IProcessor rootProcessor = new RootProcessor();
             IDictionary<string, IProcessor> sources = new Dictionary<string, IProcessor>();
-            IDictionary<string, IProcessor> sinks = new Dictionary<string, IProcessor>();
+            IDictionary<string, List<IProcessor>> sinks = new Dictionary<string, List<IProcessor>>();
             IDictionary<string, IProcessor> processors = new Dictionary<string, IProcessor>();
             IDictionary<string, IStateStore> stateStores = new Dictionary<string, IStateStore>();
             IList<string> repartitionTopics = new List<string>();
@@ -408,7 +408,7 @@ private ProcessorTopology BuildTopology(ISet<string> nodeGroup, TaskId taskId)
         private void BuildSinkNode(
             IDictionary<string, IProcessor> processors,
             IList<string> repartitionTopics,
-            IDictionary<string, IProcessor> sinks,
+            IDictionary<string, List<IProcessor>> sinks,
             ISinkNodeFactory factory,
             IProcessor processor)
         {
@@ -423,14 +423,14 @@ private void BuildSinkNode(
                 {
                     var repartitionTopic = DecorateTopic(factory.Topic);
                     repartitionTopics.Add(repartitionTopic);
-                    sinks.Add(repartitionTopic, processor);
+                    sinks.CreateListOrAdd(repartitionTopic, processor);
                     ((ISinkProcessor) processor).UseRepartitionTopic(repartitionTopic);
                 }
                 else
-                    sinks.Add(factory.Topic, processor);
+                    sinks.CreateListOrAdd(factory.Topic, processor);
             }
             else
-                sinks.Add(factory.Name, processor);
+                sinks.CreateListOrAdd(factory.Name, processor);
         }
 
         private void BuildSourceNode(

diff --git a/core/Processors/Internal/StoreChangelogReader.cs b/core/Processors/Internal/StoreChangelogReader.cs
@@ -247,13 +247,18 @@ internal bool HasRestoredToEnd(ChangelogMetadata changelogMetadata)
             long? endOffset = changelogMetadata.RestoreEndOffset;
             if (endOffset == null || endOffset == Offset.Unset || endOffset == 0)
                 return true;
+
+
+            if(changelogMetadata.CurrentOffset >= endOffset)
+                return true;
+
             if (!changelogMetadata.BufferedRecords.Any())
             {
                 var offset = restoreConsumer.Position(changelogMetadata.StoreMetadata.ChangelogTopicPartition);
                 return offset != Offset.Unset && offset >= endOffset;
             }
 
-            return changelogMetadata.CurrentOffset >= endOffset;
+            return false;
         }
 
         private void BufferedRecords(IEnumerable<ConsumeResult<byte[], byte[]>> records)

diff --git a/core/Processors/StreamThread.cs b/core/Processors/StreamThread.cs
@@ -21,28 +21,28 @@ internal class StreamThread : IThread
 
         public static string GetTaskProducerClientId(string threadClientId, TaskId taskId)
         {
-            return threadClientId + "-" + taskId + "-producer";
+            return threadClientId + "-" + taskId + "-streamiz-producer";
         }
 
         public static string GetThreadProducerClientId(string threadClientId)
         {
-            return threadClientId + "-producer";
+            return threadClientId + "-streamiz-producer";
         }
 
         public static string GetConsumerClientId(string threadClientId)
         {
-            return threadClientId + "-consumer";
+            return threadClientId + "-streamiz-consumer";
         }
 
         public static string GetRestoreConsumerClientId(string threadClientId)
         {
-            return threadClientId + "-restore-consumer";
+            return threadClientId + "-streamiz-restore-consumer";
         }
 
         // currently admin client is shared among all threads
         public static string GetSharedAdminClientId(string clientId)
         {
-            return clientId + "-admin";
+            return clientId + "-streamiz-admin";
         }
 
         internal static IThread Create(string threadId, string clientId, InternalTopologyBuilder builder,

diff --git a/core/State/InMemory/InMemoryKeyValueBytesStoreSupplier.cs b/core/State/InMemory/InMemoryKeyValueBytesStoreSupplier.cs
@@ -26,7 +26,7 @@ public InMemoryKeyValueBytesStoreSupplier(string name)
         /// <summary>
         /// Name of this state store supplier. This must be a valid Kafka topic name; valid characters are ASCII alphanumerics, '.', '_' and '-'.
         /// </summary>
-        public string Name { get; }
+        public string Name { get; set; }
 
         /// <summary>
         /// Return a new <see cref="IStateStore"/> instance.

diff --git a/core/State/InMemory/InMemoryWindowStoreSupplier.cs b/core/State/InMemory/InMemoryWindowStoreSupplier.cs
@@ -9,8 +9,6 @@ namespace Streamiz.Kafka.Net.State.InMemory
     /// </summary>
     public class InMemoryWindowStoreSupplier : IWindowBytesStoreSupplier
     {
-        private readonly TimeSpan retention;
-
         /// <summary>
         /// Constructor 
         /// </summary>
@@ -20,7 +18,7 @@ public class InMemoryWindowStoreSupplier : IWindowBytesStoreSupplier
         public InMemoryWindowStoreSupplier(string storeName, TimeSpan retention, long? size)
         {
             Name = storeName;
-            this.retention = retention;
+            Retention = (long) retention.TotalMilliseconds;
             WindowSize = size;
         }
 
@@ -32,7 +30,7 @@ public InMemoryWindowStoreSupplier(string storeName, TimeSpan retention, long? s
         /// <summary>
         /// Name of state store
         /// </summary>
-        public string Name { get; }
+        public string Name { get; set; }
 
         /// <summary>
         /// Window size of state store
@@ -42,14 +40,14 @@ public InMemoryWindowStoreSupplier(string storeName, TimeSpan retention, long? s
         /// <summary>
         /// Retention period of state store
         /// </summary>
-        public long Retention => (long)retention.TotalMilliseconds;
+        public long Retention { get; set; }
 
         /// <summary>
         /// Return a new <see cref="IWindowStore{K, V}"/> instance.
         /// </summary>
         /// <returns>Return a new <see cref="IWindowStore{K, V}"/> instance.</returns>
         public IWindowStore<Bytes, byte[]> Get()
-            => new InMemoryWindowStore(Name, retention, WindowSize.Value);
+            => new InMemoryWindowStore(Name, TimeSpan.FromMilliseconds(Retention), WindowSize.Value);
 
     }
 }
diff --git a/core/State/Internal/CompositeReadOnlyKeyValueStore.cs b/core/State/Internal/CompositeReadOnlyKeyValueStore.cs
@@ -80,7 +80,13 @@ public V Get(K key)
             IEnumerable<IReadOnlyKeyValueStore<K, V>> stores = GetAllStores();
             try
             {
-                return stores.FirstOrDefault(x => x.Get(key) != null).Get(key);
+                foreach (var store in stores)
+                {
+                    var value = store.Get(key);
+                    if (value != null)
+                        return value;
+                }
+                return default;
             }
             catch (InvalidStateStoreException e)
             {

diff --git a/core/State/RocksDb/RocksDbKeyValueBytesStoreSupplier.cs b/core/State/RocksDb/RocksDbKeyValueBytesStoreSupplier.cs
@@ -25,7 +25,7 @@ public RocksDbKeyValueBytesStoreSupplier(string name)
         /// <summary>
         /// State store name
         /// </summary>
-        public string Name { get; }
+        public string Name { get; set; }
 
         /// <summary>
         /// Build the rocksdb state store.

diff --git a/core/State/RocksDb/RocksDbWindowBytesStoreSupplier.cs b/core/State/RocksDb/RocksDbWindowBytesStoreSupplier.cs
@@ -10,7 +10,6 @@ namespace Streamiz.Kafka.Net.State.RocksDb
     /// </summary>
     public class RocksDbWindowBytesStoreSupplier : IWindowBytesStoreSupplier
     {
-        private readonly TimeSpan retention;
         private readonly long segmentInterval;
 
         /// <summary>
@@ -27,7 +26,7 @@ public RocksDbWindowBytesStoreSupplier(
             long? size)
         {
             Name = storeName;
-            this.retention = retention;
+            Retention = (long)retention.TotalMilliseconds;
             this.segmentInterval = segmentInterval;
             WindowSize = size;
         }
@@ -45,12 +44,12 @@ public RocksDbWindowBytesStoreSupplier(
         /// <summary>
         /// Retention of the state store
         /// </summary>
-        public long Retention => (long)retention.TotalMilliseconds;
+        public long Retention { get; set; }
 
         /// <summary>
         /// State store name
         /// </summary>
-        public string Name { get; }
+        public string Name { get; set; }
 
         /// <summary>
         /// Build the rocksdb state store.

diff --git a/core/State/Supplier/IStoreSupplier.cs b/core/State/Supplier/IStoreSupplier.cs
@@ -13,7 +13,7 @@ public interface IStoreSupplier<out T>
         /// Return the name of this state store supplier.
         /// This must be a valid Kafka topic name; valid characters are ASCII alphanumerics, '.', '_' and '-'.
         /// </summary>
-        string Name { get; }
+        string Name { get; set; }
 
         /// <summary>
         /// Return a new <see cref="IStateStore"/> instance of type <typeparamref name="T"/>.

diff --git a/core/State/Supplier/IWindowBytesStoreSupplier.cs b/core/State/Supplier/IWindowBytesStoreSupplier.cs
@@ -21,6 +21,6 @@ public interface IWindowBytesStoreSupplier : IStoreSupplier<IWindowStore<Bytes,
         /// <summary>
         /// The time period for which the <see cref="IWindowStore{K, V}"/> will retain historic data.
         /// </summary>
-        public long Retention { get; }
+        public long Retention { get; set; }
     }
 }
diff --git a/core/Stream/Internal/GroupedStreamAggregateBuilder.cs b/core/Stream/Internal/GroupedStreamAggregateBuilder.cs
@@ -90,7 +90,8 @@ private string Repartition(StoreBuilder storeBuilder)
         {
             if (repartitionRequired)
             {
-                (string sourceName, RepartitionNode<K,V> repartNode) = KStream<K, V>.CreateRepartitionSource(grouped.Named ?? storeBuilder.Name, grouped.Key, grouped.Value, builder);
+                string suffix = grouped.Named ?? storeBuilder.Name;
+                (string sourceName, RepartitionNode<K,V> repartNode) = KStream<K, V>.CreateRepartitionSource(suffix, grouped.Key, grouped.Value, builder);
 
                 if (repartitionNode == null || grouped.Named == null)
                     repartitionNode = repartNode;