Help Needed - IKTable into a topic

[abannan]

Hi,

Recently started using this package and I need some help. I've setup a topic which I then filter to another topic:

builder.Stream<string, string>("myTopic").Filter((key, value) => { return key == "1"; }).To("tempTopic");

This works fine. Now I want to group the message by key, reduce them and output to another topic:

builder.Stream<string, string>("tempTopic").GroupByKey().Reduce(new MyReducer()).ToStream().To("finalTopic");

This doesn't seem to be working as there are no messages in the finalTopic.

Please can you indicate where I am going wrong - basically how do I get the results from the KTable into a topic?

Thanks

[LGouellec]

Hi,

Some strange.
Did you try to write a unit test with TopologyTestDriver and this topology :

builder.Stream<string, string>("tempTopic").GroupByKey().Reduce(new MyReducer()).ToStream().To("finalTopic");

Example :

           var config = new StreamConfig<StringSerDes, StringSerDes>();
            config.ApplicationId = "test-reduce";

            var builder = new StreamBuilder();
          builder
              .Stream<string, string>("tempTopic")
              .GroupByKey()
              .Reduce(new MyReducer())
              .ToStream()
              .To("finalTopic");

            var topology = builder.Build();
            using (var driver = new TopologyTestDriver(topology, config))
            {
                var input = driver.CreateInputTopic<string, string>("tempTopic");
                var output = driver.CreateOutputTopic<string, string>("finalTopic");
                input.PipeInput("test", "1");
                input.PipeInput("test", "100");
                // todo - get value on output topic
             }

Do you have any records in output topic ?

[abannan]

Thanks for the reply. I have now tested both parts individually and they work but putting the together on the same builder doesn't work.

var config = new StreamConfig<StringSerDes, StringSerDes>()
{
    ApplicationId = "test-reduction"
};

StreamBuilder builder = new StreamBuilder();

builder
    .Stream<string, string>("myTopic")
    .Filter((key, value) =>
    {
        return key == "1";
    })
    .To("tempTopic");

builder.Stream<string, string>("tempTopic")
    .GroupByKey()
    .Reduce(new MediumEventMessageReducer())
    .ToStream()
    .To("finalTopic");

var topology = builder.Build();

using (var driver = new TopologyTestDriver(topology, config))
{
    var input = driver.CreateInputTopic<string, string>("myTopic");
    var output1 = driver.CreateInputTopic<string, string>("tempTopic");
    var output = driver.CreateOuputTopic<string, string>("finalTopic");

    input.PipeInput("1", "Once");
    input.PipeInput("2", "Once");
    input.PipeInput("1", "Twice");
    input.PipeInput("3", "Once");
    input.PipeInput("1", "Thrice");
    input.PipeInput("2", "Twice");

    var list = output.ReadKeyValueList().Select(r => KeyValuePair.Create(r.Message.Key, r.Message.Value)).ToList();

    foreach (var item in list)
    {
        Console.WriteLine(item);
    }

    Assert.IsNotNull("x");
}

Must something I'm doing?

By the way, your TopologyTestDriver doesn't have a CreateOutputTopic, it's spelt CreateOuputTopic - not sure if that's intentional

[LGouellec]

Ok,
I will check if it's a bug in the library.
Exact, it's not intentional. CreateOutputTopic was the intentional name of the method. Will rename in future minor release (seems 1.3.0)

[LGouellec]

If you run your topology with a real cluster, you haven't records in output topic too ?

[abannan]

Correct - I'm running on Kafka in a local Docker environment and nothing appears in the finalTopic

[LGouellec]

Hi @abannan,

It's a bug in restoration phase from changelog topic. Could you create an issue with label 'bug', and I will try to create a hotfix ASAP please ?
Thanks

[LGouellec]

No worries, I did for you
#103

Will release this week (1.2.1).
Branch hotfix/103.

[abannan]

Thanks for the fast response and fix - great work!

[RPalejiya]

First of all, Thank you very very much for your efforts. Really great work.

Probably I am experiencing the same issue so continuing with this thread. My stream was working in RC1 Package but not anymore.
It works for group by but if I add Aggregate on it then it hangs forever.

var StreamConfig = new StreamConfig<StringSerDes, SchemaAvroSerDes<GenericRecord>>();
var stream = streamBuilder.Stream<string, GenericRecord>("TOPIC");
var grouped = stream.GroupBy(
                            //New Key Slection Function
                            (k, v) =>
                            {
                                StringBuilder s = new();
                                bool first = true;
                                foreach (var c in AggregateColumns)
                                {
                                    if (!first) s.Append("|");
                                    v.TryGetValue(c, out var f);
                                    s.Append(f);
                                    first = false;
                                }

                                Console.WriteLine($"Grouped {s}");
                                return s.ToString();
                            }
                        );

// THIS PART CAUSES THE PROBLEM
IKTable<string, double> kpiTable = grouped.Aggregate<double, DoubleSerDes>(
                                                    () => (double)0,  //Initializer
                                                    (k, v, sumvalue) //Aggregator
                                                        =>
                                                    {
                                                        double total = 0;
                                                        Console.WriteLine($"Totaled {total}");
                                                        return total;
                                                    });


var topology  = streamBuilder.Build();
var kafkaStream  = new KafkaStream(topology, StreamConfig);
kafkaStream.StartAsync().Wait();

[LGouellec]

Hi @RPalejiya,
Yes exactly, there is a bug when you use statefull processor with changelog topic (activate by default since 1.2.0) and restoration phase.
This bug was fix in future version (1.2.1) based on this branch and will be available before ending this week.
Sorry for the inconvenience

[LGouellec]

Release 1.2.1 is available : https://github.com/LGouellec/kafka-streams-dotnet/releases/tag/v1.2.1

[abannan]

All works now - thanks for the quick response.