Questions about how to use Kafka Streams with Confluent.Kafka Headers

[ncupertino]

Hello,

I'm trying to produce an event on an output topic with the Confluent.Kafka Headers object using Message and this error is occurring:

Streamiz.Kafka.Net.Processors.Internal.TaskManager[0]
Failed to process stream task 0-0 due to the following error:
Streamiz.Kafka.Net.Errors.StreamsException: stream-task[0|0]|processor[KSTREAM-SINK-0000000004]- The value serdes (Streamiz.Kafka.Net.SerDes.StringSerDes) is not compatible with the actual value ( Confluent.Kafka.Message`2[System.String,System.String]) for this processor. Change the default value serdes in StreamConfig or provide correct Serdes via method parameters(using the DSL)

My code looks like this:

I declared the config
var config = new StreamConfig<StringSerDes, StringSerDes>

And the implementation:
builder.Stream<string, string>("topic-in", new StringSerDes(), new StringSerDes())
.Peek((k, v) => Console.WriteLine($" K: {k} V: {v}"))
.MapValues((key, value) =>
{

             var headers = new Headers();
             headers.Add("headerKey", new byte[] { 0x01, 0x02, 0x03 });


             var message = new Message<string, string>
             {
                 Key = null, // Key can be null
                 Value = value,
                 Headers = headers
             };

    


             return message;
         })
     .To("topic-out");

Is there any way?

[LGouellec]

Hi @ncupertino ,

You can add or remove headers with StreamizMetadata.GetCurrentHeadersMetadata() inside a lambda function in Map(..), MapValues(...) or etc ...

            var builder = new StreamBuilder();

            builder.Stream<string, string>("topic-in", new StringSerDes(), new StringSerDes())
                .Peek((k, v) => Console.WriteLine($" K: {k} V: {v}"))
                .Map((key, value) => {

                    StreamizMetadata
                        .GetCurrentHeadersMetadata()
                        .Add("headerKey", new byte[] { 0x01, 0x02, 0x03 });

                    return KeyValuePair.Create<string, string>(null, value);
                })
                .To("topic-out");

[ncupertino]

Hello @LGouellec !Thank you for your help. However, unable to implement. GetCurrent Headers Metadata return null and consequently an error occurs. How can I resolve this issue?

[LGouellec]

@ncupertino
You have to enable this feature via :

config.FollowMetadata = true;

https://github.com/LGouellec/kafka-streams-dotnet/blob/43d8a6e88f1480212c4d1996603687c622ed5f6f/core/StreamConfig.cs#L2371

[ncupertino]

It worked!
However i dont user to publish message KeyValuePair.Create<string, string>(null, value); , i only returned the processed message and it worked too.
Code:

builder.Stream<string, string>("topicIn", new StringSerDes(), new StringSerDes())
            .Peek((k, v) => Console.WriteLine($" K: {k} V: {v}"))
            .MapValues((key, value) =>
            {
                Console.WriteLine($" ToUpper ${value.ToUpper()}");

                var header = "Worker2";
                var header2 = "Worked2";

                StreamizMetadata
                .GetCurrentHeadersMetadata()
                .Add("headerKey", Encoding.ASCII.GetBytes(header));

                StreamizMetadata
                .GetCurrentHeadersMetadata()
                .Add("headerKey2", Encoding.ASCII.GetBytes(header2));

                return value;

                

            })
        .To("topicOut");

What is the benefit of using KeyValuePair.Create<string, string>?

[LGouellec]

MapValues(..) permit to modify only the value whereas Map(..) permit to modify the key and the value at the same time