Kafka config cleanup

Introduction

In order to clean up topics on Kafka that have configs that are not supposed to be there, a small app has been developed: kafka-config-cleanup

Why do I need to remove those properties?

here are some configs that were marked as deprecated somewhere in the Kafka 3 line. These configs have been removed in Kafka 4. This means that, if those configs are explicitly set on topics, you cannot make any modification to the topic until those specific configs are removed.

Axual Platform allows for one of these properties to be set explicitly, which is message.timestamp.difference.max.ms, hence the need to explicitly remove it.

This property is used in Kafka to determine whether a message timestamp is valid, in case the timestamp type is CreateTime. The default on the Axual Kafka clusters is LogAppendTime, hence there will not be any impact of the removal of this property.

Application details

Note that any platform operator is of course free to use a client or Kafka administration tool of their choice to remove this config.

This is a Spring Boot application using Spring Boot starter version 4.0.2 with Spring Kafka. Because of this, configuration is rather easy and all default Spring Kafka configuration options are available to the user, for example

spring:
  kafka:
    bootstrap-servers: axual-kafka-bootstrap:9092
    security:
      protocol: SASL_SSL
    ssl:
      trust-store-certificates: |
        -----BEGIN CERTIFICATE-----
        ...
        -----END CERTIFICATE-----
      trust-store-type: "PEM"
    properties:
      sasl.jaas.config: 'org.apache.kafka.common.security.scram.ScramLoginModule required username="config-cleanup" password="password";'
      sasl.mechanism: SCRAM-SHA-512

This app can be configured to delete certain configuration keys on topics that match a pattern. Specific app config is as follows

topic:
  dryrun: true
  pattern: "^.*$"
  include-internal-topics: false
  mode: delete
  config:
    delete:
      - "message.timestamp.difference.max.ms"
      - "message.downconversion.enable"
      - "message.format.version"
    set: {}

Config Description Values

Config	Description	Values
`topic.dryrun`	Indicate whether to actually apply the changes or only log them.	`true`, `false`. Default: `true`
`topic.pattern`	Regular expression pattern for topic selection. Only apply changes to topics that match this pattern.	Regular expression string. Default: `"^.*$"`
`topic.include-internal-topics`	Whether to include internal topics in the selection. These are usually topics starting with `_` and would generally not be required to include.	`true`, `false`. Default: `false`
`topic.mode`	Specifies if the application should delete or set configurations.	`delete`, `set`. Default: `delete`
`topic.config.delete`	List of configuration keys to remove from the selected topics.	List of strings. Default: `["message.timestamp.difference.max.ms", "message.downconversion.enable", "message.format.version"]`
`topic.config.set`	Map of configuration keys and values to set on the selected topics.	Map of configuration `key: value`. Default: `{}`

topic.dryrun

Indicate whether to actually apply the changes or only log them.

true, false. Default: true

topic.pattern

Regular expression pattern for topic selection. Only apply changes to topics that match this pattern.

Regular expression string. Default: "^.*$"

topic.include-internal-topics

Whether to include internal topics in the selection. These are usually topics starting with _ and would generally not be required to include.

true, false. Default: false

topic.mode

Specifies if the application should delete or set configurations.

delete, set. Default: delete

topic.config.delete

List of configuration keys to remove from the selected topics.

List of strings. Default: ["message.timestamp.difference.max.ms", "message.downconversion.enable", "message.format.version"]

topic.config.set

Map of configuration keys and values to set on the selected topics.

Map of configuration key: value. Default: {}

Usage

The application uses a Kafka AdminClient to do its job. The User permissions required are:

Describe on topics
AlterConfigs on topics
DescribeConfigs on topics. Note that this permission is implicitly granted through giving AlterConfigs permissions.

The app can be used in “dryrun” mode to view in the logs which topics would be modified with which modification (deletion of config properties most likely). In order to remove the Kafka 4 incompatible configurations that could have been set once, configure it with the yaml above.

The logs will indicate what would happen if dryrun was disabled.

2026-01-29 15:36:11,357 INFO  [main] com.axual.iris.admin.TopicConfigHandler: Altering configs for
topic: acceptance-dta-tst-daniel1
    DELETE message.timestamp.difference.max.ms
2026-01-29 15:36:11,357 INFO  [main] com.axual.iris.admin.TopicConfigHandler: Dryrun mode enabled, not applying config changes. Exiting

Running the cleanup app as a Kubernetes job

This image can be run as a simple Kubernetes job, feeding it a configmap and secrets for any configuration it might need. An example is given below

---
apiVersion: batch/v1
kind: Job
metadata:
  name: kafka-config-cleanup
spec:
  ttlSecondsAfterFinished: 10
  backoffLimit: 0
  template:
    spec:
      containers:
        - name: kafka-config-cleanup
          image: registry.axual.io/axual/iris/kafka-config-cleanup:0.2.3
          imagePullPolicy: Always
          volumeMounts:
            - name: config-volume
              mountPath: /config
          env:
            - name: "SPRING_CONFIG_LOCATION"
              value: "file:///config/application.yaml"
            - name: "TRUSTSTORE"
              valueFrom:
                secretKeyRef:
                  key: "tls.crt"
                  name: "<secret with broker issuers / certs>"
            - name: "SASL_JAAS_CONFIG"
              valueFrom:
                secretKeyRef:
                  name: "kafka-config-cleaner" # Secret containing the full jaas config.
                  key: "sasl.jaas.config"
      imagePullSecrets:
        - name: axualdockercred
      volumes:
        - name: config-volume
          configMap:
            name: kafka-config-cleanup-config
      restartPolicy: Never
---
apiVersion: v1
kind: ConfigMap
metadata:
  name: kafka-config-cleanup-config
data:
  application.yaml: |
    topic:
      dryrun: true
      include-internal-topics: false
      mode: delete
      pattern: "^.*$"
      config:
        delete:
          - "message.timestamp.difference.max.ms"
          - "message.downconversion.enable"
          - "message.format.version"
    spring:
      kafka:
        bootstrap-servers: axual-kafka-bootstrap:9093
        security:
          protocol: SASL_SSL
        ssl:
          trust-store-certificates: ${TRUSTSTORE}
          trust-store-type: "PEM"
        properties:
          sasl.jaas.config: ${SASL_JAAS_CONFIG}
          sasl.mechanism: SCRAM-SHA-512

This will run a one-off job that removes the properties message.timestamp.difference.max.ms, message.downconversion.enable and message.format.version from any topics matching regex "^.*$" - excluding internal topics (as can be seen in config). It will dry run, indicating what it will do when chosen to actually execute. The application logs will give a clear indication of the modifications that the app will perform when dryrun is turned off (as mentioned above).