Kafka Streams Internal Topic Naming

4
votes

According to the documentation (https://docs.confluent.io/current/streams/developer-guide/manage-topics.html#internal-topics), internal topics follow the naming convention <application.id>-<operatorName>-<suffix>.

Some examples we have are:

testapplication-KSTREAM-REDUCE-STATE-STORE-0000000008-repartition  
testapplication-KSTREAM-REDUCE-STATE-STORE-0000000027-repartition

Does anyone know how the integer are determined?

Unfortunately our security requirements do not allow for us to create topics with our applications and need to be setup ahead of time. I am trying to determine if these topic names will be consistent.

5
apache-kafkaapache-kafka-streams
Why did you change your question? This makes existing answers void... You should have rather asked a new question. - Matthias J. Sax
It an implementation detail and you should not rely how those name are generated. Internally, it's just a counter that is increase each time a new name is generated. - Matthias J. Sax
If this is an issue, you can also name certain operators/stores to set the names in your code. - Matthias J. Sax
How do you do that? We are using the DSL api and see no way for the reduce operation to name the topic. If you can write an answer describing that it would be great! - Chris
All stateful operator have overload to pass in optional argument like Materialized or Joined that allow to specify names: If you could read the docs, that would be great: kafka.apache.org/20/javadoc/org/apache/kafka/streams/kstream/… - Matthias J. Sax

5 Answers

2
votes

Usually the intermediate topic names are constructed with following convention:

<ApplicationId>-<operator name>-<suffix>

Suffix value can be either "changelog" or "repartition"

Based on the operator, it uses one of the suffix. Here is an example:

testapplication-aggregate-repartition

testapplication-aggregate-changelog

1
votes

To answer your main question about the integer, that's what I found in docs:

The number is a globally incrementing number that represents the operator’s order in the topology. The generated number is prefixed with a varying number of “0”s to create a string that is consistently 10 characters long.

This is quite important aspect of kafka streams DSL and can lead to some problems if you change a topology. It's a good practice to name your stateful operator.

More information you can find in dsl-topology-naming article

0
votes

Have you seen these commands to set ACL for Streams internal topics. I believe they are introduced as part of Kafka v2.x.x (confluent doc)

   # Allow Streams to manage its own internal topics and consumer groups:
   bin/kafka-acls ... --add --allow-principal User:team1 --operation All --resource- 
   pattern-type prefixed --topic team1-streams-app1 --group team1-streams-app1

So you just need to know the steams application.id, which is the prefix of all internal topics.

I believe as you will give permission ALL, that would allow their creation as well.

0
votes

To answer your question about topic names being consistent, in my experience, they have been consistent between executions of the application, however, if you modify the order of, add, or delete any joins or reduces within your logic, the topic names may change.

-1
votes

Integers are internally generated.

You can find it documented here:

https://docs.confluent.io/current/streams/javadocs/index.html

Under groupBy method descriptions it says:

Because a new key is selected, an internal repartitioning topic will be created in Kafka. This topic will be named "${applicationId}-XXX-repartition", where "applicationId" is user-specified in StreamsConfig via parameter APPLICATION_ID_CONFIG, "XXX" is an internally generated name, and "-repartition" is a fixed suffix. You can retrieve all generated internal topic names via Topology.describe().