Question: How to overwrite prefix of internal topics (application id fails)

[DatStorm]

Hey.

First of thanks for a lot of great work!

Description

We have a problem with our internal topics they do not seem to follow the pattern described here: https://docs.confluent.io/platform/current/streams/developer-guide/manage-topics.html

We are using Confluent cloud and to ensure no clashes between topics we have created an application id prefix that should be applied to all internal topics but this does not seem to be the case.

How to reproduce

1. Start the example application
2. Look at the initializing logs and see that internal stream topics are not prefixed with application id but with KSTREAM-MAPVALUES-ASYNC-0000000008-request

Checklist

More info can be provided if needed.

Please provide the following information:

• [X ] A complete (i.e. we can run it), minimal program demonstrating the problem. No need to supply a project file.
• [ ] A code snippet with your topology builder (ex: builder.Stream<string, string>("topic").to("an-another-topic");)
• [ ] Streamiz.Kafka.Net nuget version.
• [ ] Apache Kafka version.
• [ ] Client configuration.
• [ ] Operating system.
• [ ] Provide logs (with in debug mode (log4net and StreamConfig.Debug) as necessary in configuration).
• [ ] Critical issue.

[LGouellec]

Hi @DatStorm,

Can you share your topology please ?

Best regards,

[DatStorm]

Sure.

Our config

       var config = new StreamConfig<StringSerDes, StringSerDes>
            {

                ClientId = "<client_id>",
                ApplicationId = "<application_id>", // <--- Not being prefixed.
                Debug = "security,broker,topic,msg,consumer,cgrp,topic,fetch",
                BootstrapServers = "confluent.cloud....",
               // login removed...
                SaslMechanism = SaslMechanism.Plain,
                Logger = LoggerFactory.Create(builder => builder.AddLog4Net()),
                MetricsRecording = MetricsRecordingLevel.DEBUG
            };

        return config;

Our topology:

  var builder = new StreamBuilder();

        // re key stream 
        builder.Stream<string, string>(topicsConfig.SupplierInputTopicName())
            .Map((k, v) => KeyValuePair.Create(v, v))
            .To(topicsConfig.SupplierOutputTopicName());

        // transform stream 
        builder.Stream<string, string>(topicsConfig.SupplierOutputTopicName())
            .Peek((k, v) => _logger.LogInformation("key={k} value={v}", k,v))
            .MapValuesAsync<SupplierEventModel>(async (record, _) => await TransformationToSupplierEvent(record))
            .To(topicsConfig.SupplierTransformedTopic());

        var topology = builder.Build();

        return new KafkaStream(topology, streamConfig);

[LGouellec]

@DatStorm ,

I tried with this topology and the internal request and response topics are correctly prefixed with the application.id

      var config = new StreamConfig<StringSerDes, StringSerDes>();
            config.ApplicationId = "test-app";
            config.BootstrapServers = "localhost:9092";

            StreamBuilder builder = new StreamBuilder();

            builder.Stream<string, string>("inputs")
                .Peek((k, v) => logger.LogInformation("key={k} value={v}", k,v))
                .MapValuesAsync(async (record, _) => await Task.FromResult(record.Value.ToUpper()))
                .To("output");

            Topology t = builder.Build();
            KafkaStream stream1 = new KafkaStream(t, config);

            Console.CancelKeyPress += (_, _) => stream1.Dispose();

            await stream1.StartAsync();

[DatStorm]

I just ran it locally and you are right the topics created are in fact prefixed with the application id but the topology printed in the log (in the beginning) does not have them and that is a bit confusing.

[LGouellec]

The topology printed at the beginning is agnostic to the application id. Same for the others internal topics (changelog and repartition).

[DatStorm]

I was not aware. So for the misunderstanding.

The issue can be closed.