{"id":2283,"date":"2022-07-13T14:21:34","date_gmt":"2022-07-13T14:21:34","guid":{"rendered":"https:\/\/blog.samarthya.me\/wps\/?p=2283"},"modified":"2022-07-13T14:21:36","modified_gmt":"2022-07-13T14:21:36","slug":"naming-convention-policy-for-kafka-topics","status":"publish","type":"post","link":"https:\/\/blog.samarthya.me\/wps\/2022\/07\/13\/naming-convention-policy-for-kafka-topics\/","title":{"rendered":"Naming Convention Policy for Kafka topics"},"content":{"rendered":"\n

Creating a\u00a0Topic<\/strong>\u00a0in a kafka-cluster is easy and well documented for the `kafka-topics.sh` or even the official API documentation.<\/p>\n\n\n\n

 bin\/kafka-topics.sh --help<\/code><\/pre>\n\n\n\n
The complexity arises, when you are trying to enforce a standard way of defining topic naming. There are many ways to identify the right convention<\/strong> based on your need, but to enforce such a topic convention while you are creating one is explained in this 5 steps blog.<\/p>\n\n\n\n
There is no right convention<\/em> it is always determined based on what your business needs.<\/p><\/blockquote>\n\n\n\n
For my example, I wish to define to define a topic convention that follows the semantics as under<\/p>\n\n\n\n
<organizationname>.<productname><\/code><\/pre>\n\n\n\n
It is simple enough to get started, and can be easily extended as you will observe as you follow along.<\/p>\n\n\n\n
From the official documentation<\/a>, if you wish to define a custom topic policy creation you will have to define property as under<\/p>\n\n\n\n
Properties files<\/p>\n\n\n\n
create.topic.policy.class.name=mypackage.className<\/code><\/pre>\n\n\n\n
The className should implement the interface <\/p>\n\n\n\n
 org.apache.kafka.server.policy.CreateTopicPolicy<\/code><\/pre>\n\n\n\n
Step 1: Building Project<\/h2>\n\n\n\n
With these two building blocks let’s define a maven project<\/p>\n\n\n\n
\"\"\/
Project Space<\/figcaption><\/figure>\n\n\n\n
Step 2: Define the dependency<\/h2>\n\n\n\n
Let’s define a package `me.samarthya` and also add dependency of the kafka-clients in the `pom.xml`<\/p>\n\n\n\n
 <dependency>
   <groupId>org.apache.kafka<\/groupId>
   <artifactId>kafka-clients<\/artifactId>
   <version>3.2.0<\/version>
   <scope>compile<\/scope>
<\/dependency><\/code><\/pre>\n\n\n\n
Step 3: Implementation<\/h2>\n\n\n\n
Lets define the main class Topic Policy as under<\/p>\n\n\n\n
public class TopicPolicy implements CreateTopicPolicy {
    private final Logger logger = Logger.getLogger(TopicPolicy.class.toString());

    private final static String TopicPattern = \"\\\\w+\\\\.{1}\\\\w+\";

    @Override
    public void validate(RequestMetadata requestMetadata) throws PolicyViolationException {
        StringBuilder bd = new StringBuilder().append(\" Topic Name=\").append(requestMetadata.topic());
        logger.info(bd.toString());
        if ( requestMetadata.topic().isEmpty() || !Pattern.matches(TopicPattern, requestMetadata.topic())) {
            throw new PolicyViolationException(\"Topic name \" + requestMetadata.topic() + \" should match the pattern \" + TopicPattern);
        }
    }

    @Override
    public void close() throws Exception {
        logger.info(\" Close & release.\");
    }

    @Override
    public void configure(Map<String, ?> configs) {
        if (configs != null) {
            for( String k: configs.keySet()) {
                logger.info(configs.get(k).toString());
            }
        }
    }
}<\/code><\/pre>\n\n\n\n
With the class defined the main thing to observe is, the TopicPattern <\/strong>that has been defined as the format that will be matched for the name, and if it is not found a PolicyViolationException will be thrown.<\/p>\n\n\n\n
Step 4: Repeat for each broker in the cluster<\/strong><\/h2>\n\n\n\n
Package the jar and it has to be placed under the `lib` folder of the kafka (classpath).  <\/strong><\/p>\n\n\n\n
PowerShell<\/p>\n\n\n\n
  4 -rw-r--r--. 1 vagrant vagrant     3881 Jul 12 06:28 topic-policy-1.0-SNAPSHOT.jar<\/code><\/pre>\n\n\n\n
also, in the `server.properties` you can define two properties<\/p>\n\n\n\n
create.topic.policy.class.name=me.samarthya.TopicPolicy\nauto.create.topics.enable=false<\/code><\/pre>\n\n\n\n
Restart your cluster.<\/p>\n\n\n\n
Step 5: Test your `Topics`<\/h2>\n\n\n\n
Let’s go back to the kafka binary folder (local machine) and issue the topic creation command again<\/p>\n\n\n\n
 bin\/kafka-topics.sh --bootstrap-server mybroker.test:9092  --topic invalid_topic --create<\/code><\/pre>\n\n\n\n
If the jar has been loaded successfully you should see an error reported as below<\/p>\n\n\n\n
WARNING: Due to limitations in metric names, topics with a period ('.') or underscore ('_') could collide. To avoid issues it is best to use either, but not both.
Error while executing topic command : Topic name invalid_topic should match the pattern \\w+\\.{1}\\w+
[2022-07-13 09:49:21,805] ERROR org.apache.kafka.common.errors.PolicyViolationException: Topic name invalid_topic should match the pattern \\w+\\.{1}\\w+
 (kafka.admin.TopicCommand$)<\/code><\/pre>\n\n\n\n
You can modify the pattern now as per your convenience and re-deploy the jar to check the new custom topic policies.<\/p>\n\n\n\n
Example:<\/h3>\n\n\n\n
bin\/kafka-topics.sh --bootstrap-server mybroker.test.test:9092  --topic invalid.valid --create<\/code><\/pre>\n\n\n\n
WARNING: Due to limitations in metric names, topics with a period ('.') or underscore ('_') could collide. To avoid issues it is best to use either, but not both.\nCreated topic invalid.valid<\/code><\/pre>\n\n\n\n
Note:<\/h2>\n\n\n\n
Since the auto-topic creation has been disabled if you try and create an invalid topic through producer it will not work (below).<\/p>\n\n\n\n
PowerShell<\/p>\n\n\n\n
 bin\/kafka-console-producer.sh --bootstrap-server mybroker.test:9092 --topic test<\/code><\/pre>\n\n\n\n
Will result in following error<\/p>\n\n\n\n
[2022-07-13 09:54:21,196] WARN [Producer clientId=console-producer] Error while fetching metadata with correlation id 4 : {test=UNKNOWN_TOPIC_OR_PARTITION} (org.apache.kafka.clients.NetworkClient)<\/code><\/pre>\n\n\n\n
For an existing topic `invalid.valid` it should work<\/p>\n\n\n\n
bin\/kafka-console-producer.sh --bootstrap-server mybrokers.test:9092 --topic invalid.valid<\/code><\/pre>\n","protected":false},"excerpt":{"rendered":"
Creating a\u00a0Topic\u00a0in a kafka-cluster is easy and well documented for the `kafka-topics.sh` or even the official API documentation. The complexity arises, when you are trying to enforce a standard way of defining topic naming. There are many ways to identify the right convention based on your need, but to enforce such a topic convention while you are […]<\/p>\n","protected":false},"author":2,"featured_media":2284,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"image","meta":{"_exactmetrics_skip_tracking":false,"_exactmetrics_sitenote_active":false,"_exactmetrics_sitenote_note":"","_exactmetrics_sitenote_category":0,"footnotes":""},"categories":[120,34,239],"tags":[225,257],"class_list":["post-2283","post","type-post","status-publish","format-image","has-post-thumbnail","hentry","category-java","category-technical","category-technical-2","tag-kafka","tag-topics","post_format-post-format-image"],"_links":{"self":[{"href":"https:\/\/blog.samarthya.me\/wps\/wp-json\/wp\/v2\/posts\/2283","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/blog.samarthya.me\/wps\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/blog.samarthya.me\/wps\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/blog.samarthya.me\/wps\/wp-json\/wp\/v2\/users\/2"}],"replies":[{"embeddable":true,"href":"https:\/\/blog.samarthya.me\/wps\/wp-json\/wp\/v2\/comments?post=2283"}],"version-history":[{"count":3,"href":"https:\/\/blog.samarthya.me\/wps\/wp-json\/wp\/v2\/posts\/2283\/revisions"}],"predecessor-version":[{"id":2287,"href":"https:\/\/blog.samarthya.me\/wps\/wp-json\/wp\/v2\/posts\/2283\/revisions\/2287"}],"wp:featuredmedia":[{"embeddable":true,"href":"https:\/\/blog.samarthya.me\/wps\/wp-json\/wp\/v2\/media\/2284"}],"wp:attachment":[{"href":"https:\/\/blog.samarthya.me\/wps\/wp-json\/wp\/v2\/media?parent=2283"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/blog.samarthya.me\/wps\/wp-json\/wp\/v2\/categories?post=2283"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/blog.samarthya.me\/wps\/wp-json\/wp\/v2\/tags?post=2283"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}