Grouping operation issue in topology

[oesebus]

IKStream<string, XtBridgeEvent> KS0 = builder.Stream("XtBridgeEvent", new StringSerDes(), new XtBridgeEventSerDes()); IKStream<string, XtMediaAssetOrchestratorCommand> KS1 = builder.Stream("XtMediaAssetOrchestratorCommand", new StringSerDes(), new XtMediaAssetOrchestratorCommandSerDes()); IKStream<string, XtMediaAssetOrchestratorEvent> KS2 = builder.Stream("XtMediaAssetOrchestratorEvent", new StringSerDes(), new XtMediaAssetOrchestratorEventSerDes()); IKStream<string, XtBridgeCommand> KS3 = builder.Stream("XtBridgeCommand", new StringSerDes(), new XtBridgeCommandSerDes()); IKStream<string, RecorderGroupQueryCommand> KS4 = builder.Stream("RecorderGroupQueryCommand", new StringSerDes(), new RecorderGroupQueryCommandSerDes());
Filtering Operations

var KS0Keyed = KS0.Filter((_, v) => v.SpecificCase is not XtBridgeEvent.SpecificOneofCase.None).SelectKey((_, v) => v.Headers.DiscussionId).MapValues((v) => StoryBuilder.Map(v)); var KS1Keyed = KS1.Filter((_, v) => v.SpecificCase is not XtMediaAssetOrchestratorCommand.SpecificOneofCase.None).SelectKey((_, v) => v.Headers.DiscussionId).MapValues((v) => StoryBuilder.Map(v)); var KS2Keyed = KS2.Filter((_, v) => v.SpecificCase is not XtMediaAssetOrchestratorEvent.SpecificOneofCase.None).SelectKey((_, v) => v.Headers.DiscussionId).MapValues((v) => StoryBuilder.Map(v)); var KS3Keyed = KS3.Filter((_, v) => v.SpecificCase is not XtBridgeCommand.SpecificOneofCase.None).SelectKey((_, v) => v.Headers.DiscussionId).MapValues((v) => StoryBuilder.Map(v)); var KS4Keyed = KS4.Filter((_, v) => v.SpecificCase is not RecorderGroupQueryCommand.SpecificOneofCase.None).SelectKey((_, v) => v.Headers.DiscussionId).MapValues((v) => StoryBuilder.Map(v));

Merging Operation + grouping + Count
var mergedRawDiscussionKTable = KS0Keyed.Merge(KS1Keyed) .Merge(KS2Keyed) .Merge(KS3Keyed) .Merge(KS4Keyed) .GroupBy<string,StringSerDes>((k,_) => k) .Count(InMemory<string, long>.As("DiscussionCountState")); mergedRawDiscussionKTable.ToStream().Peek((k, v) => Console.WriteLine($" Key {k} | Timestamp : {v}"));

Error =>
Inner Exception handling => Streamiz.Kafka.Net.Errors.StreamsException: stream-task[0|0]|processor[KSTREAM-SINK-0000000026]- The value serdes (Streamiz.Kafka.Net.SerDes.StringSerDes) is not compatible to the actual value (Oesebus.EVS.KafkaService.Application.Core.Models.StoryBuilder) for this processor. Change the default value serdes in StreamConfig or provide correct Serdes via method parameters(using the DSL)
at Streamiz.Kafka.Net.Processors.SinkProcessor2.Process(K key, V value) at Streamiz.Kafka.Net.Processors.AbstractProcessor2.Forward(K key, V value)
at Streamiz.Kafka.Net.Processors.KStreamFilterProcessor`2.Process(K key, V value)

Any idea why i receive such error ?

Thanks a lot for your help

[LGouellec]

Hi @oesebus ,

Regarding your topology, I suppose that KS0Keyed, KS1Keyed, KS2Keyed, KS3Keyed and KS4Keyed are a IKStream<string, StoryBuilder>, you had change the key before grouping so a few repartitions topics are required.

But the groupBy operation are not the correct value serdes, could you try this one ?

var mergedRawDiscussionKTable = 
  KS0Keyed
   .Merge(KS1Keyed)
   .Merge(KS2Keyed)
   .Merge(KS3Keyed)
   .Merge(KS4Keyed)
   .GroupByKey<StringSerDes, StoryBuilderSerDes>()
   .Count(InMemory<string, long>.As("DiscussionCountState")); 

mergedRawDiscussionKTable.ToStream().Peek((k, v) => Console.WriteLine($" Key {k} | Timestamp : {v}"));

Regards,

[oesebus]

Ok so i did repartition as show below :

Toplogy :
var mergedRawDiscussionKTable = KS0Keyed.Merge(KS1Keyed) .Merge(KS2Keyed) .Merge(KS3Keyed) .Merge(KS4Keyed) .Repartition(Repartitioned<string, StoryBuilder>.NumberOfPartitions(4)) .GroupByKey<StringSerDes, StorySerdes>() .Aggregate(() => new DiscussionSummary(), (_, v, old) => old.Aggregate(v));

But now i am facing this issue :
Inner Exception handling => Confluent.Kafka.KafkaException: Local: Erroneous state at Confluent.Kafka.Impl.SafeKafkaHandle.AssignImpl(IEnumerable1 partitions, Func3 assignMethodErr, Func3 assignMethodError)
at Confluent.Kafka.Impl.SafeKafkaHandle.Assign(IEnumerable1 partitions) at Confluent.Kafka.Consumer2.Unassign()
at Confluent.Kafka.Consumer2.RebalanceCallback(IntPtr rk, ErrorCode err, IntPtr partitionsPtr, IntPtr opaque) at Confluent.Kafka.Impl.NativeMethods.NativeMethods.rd_kafka_consumer_poll(IntPtr rk, IntPtr timeout_ms) at Confluent.Kafka.Consumer2.Consume(Int32 millisecondsTimeout)
at Confluent.Kafka.Consumer2.Consume(TimeSpan timeout) at Streamiz.Kafka.Net.Crosscutting.KafkaExtensions.ConsumeRecords[K,V](IConsumer2 consumer, TimeSpan timeout, Int64 maxRecords)
at Streamiz.Kafka.Net.Processors.StreamThread.PollRequest(TimeSpan ts)
at Streamiz.Kafka.Net.Processors.StreamThread.<>c__DisplayClass57_0.b__0()
at Streamiz.Kafka.Net.Crosscutting.ActionHelper.MeasureLatency(Action action)
at Streamiz.Kafka.Net.Processors.StreamThread.Run()
`

Thanks for your feedback,

Ismael

[LGouellec]

You have to set the serdes explicitly in repartition level

[oesebus]

var mergedRawDiscussionKTable = KS0Keyed.Merge(KS1Keyed) .Merge(KS2Keyed) .Merge(KS3Keyed) .Merge(KS4Keyed) .Repartition(Repartitioned<string, StoryBuilder>.NumberOfPartitions(6).WithKeySerdes(new StringSerDes()).WithValueSerdes(new StorySerdes())) .GroupByKey<StringSerDes, StorySerdes>() .Aggregate(() => new DiscussionSummary(), (_, v, old) => old.Aggregate(v));

But same problem and the error seems to be related to partition assignment during rebalance

Inner Exception handling => Confluent.Kafka.KafkaException: Local: Erroneous state
at Confluent.Kafka.Impl.SafeKafkaHandle.AssignImpl(IEnumerable1 partitions, Func3 assignMethodErr, Func3 assignMethodError) at Confluent.Kafka.Impl.SafeKafkaHandle.Assign(IEnumerable1 partitions)
at Confluent.Kafka.Consumer2.Unassign() at Confluent.Kafka.Consumer2.RebalanceCallback(IntPtr rk, ErrorCode err, IntPtr partitionsPtr, IntPtr opaque)
at Confluent.Kafka.Impl.NativeMethods.NativeMethods.rd_kafka_consumer_poll(IntPtr rk, IntPtr timeout_ms)
at Confluent.Kafka.Consumer2.Consume(Int32 millisecondsTimeout) at Confluent.Kafka.Consumer2.Consume(TimeSpan timeout)
at Streamiz.Kafka.Net.Crosscutting.KafkaExtensions.ConsumeRecords[K,V](IConsumer`2 consumer, TimeSpan timeout, Int64 maxRecords)
at Streamiz.Kafka.Net.Processors.StreamThread.PollRequest(TimeSpan ts)
at Streamiz.Kafka.Net.Processors.StreamThread.<>c__DisplayClass57_0.b__0()
at Streamiz.Kafka.Net.Crosscutting.ActionHelper.MeasureLatency(Action action)
at Streamiz.Kafka.Net.Processors.StreamThread.Run()

Any help would be appreciate ^^

Ismael

[LGouellec]

As the #163 said, when an exception is throw during partition assigmnent, we lost the source of the exception, and we have just this one :

Inner Exception handling => Confluent.Kafka.KafkaException: Local: Erroneous state
at Confluent.Kafka.Impl.SafeKafkaHandle.AssignImpl(IEnumerable1 partitions, Func3 assignMethodErr, Func3 assignMethodError) at Confluent.Kafka.Impl.SafeKafkaHandle.Assign(IEnumerable1 partitions)

Could you clone my repository and run in debug your topology to know the source of exception ?

Thanks

[LGouellec]

Link : #163