Skip to main content
Quick navigation

Alter Topic Config Policy

Introduction

The alter topic config policy interceptor will impose limits on configuration changes to ensure that any configuration changed in the topic adhere to the configured specification.

The full list of Kafka configurations that this interceptor protects is:

  • retention.ms
  • retention.bytes
  • segment.ms
  • segment.bytes
  • segment.jitter.ms
  • flush.messages
  • flush.ms
  • max.message.bytes
  • min.insync.replicas
  • cleanup.policy
  • unclean.leader.election.enable

What happens when sending an invalid request:

Any request that doesn't match the interceptor's configuration will be blocked and return the corresponding error message.

For example:

you want to change the configuration retention.ms = 10000. But the interceptor is being configured minRetentionMs=60000.

When you send that request to the cluster, the following error is returned:

org.apache.kafka.common.errors.PolicyViolationException: 
Request parameters do not satisfy the configured policy. retention.ms is '1', must not be less than '10'

Configuration

keytypedefaultdescription
topicString.*Topics that match this regex will have the interceptor applied
blacklistBlackListBlacklist of properties which cannot be changed
retentionMsLongConfiguration for retention.ms
retentionBytesLongConfiguration for retention.bytes
segmentMsLongConfiguration for segment.ms
segmentBytesIntegerConfiguration for segment.bytes
segmentJitterMsLongConfiguration for segment.jitter.ms
flushMessagesLongConfiguration for flush.messages
flushMsLongConfiguration for flush.ms
maxMessageBytesIntegerConfiguration for max.message.bytes
minInsyncReplicasIntegerConfiguration for min.insync.replicas
cleanupPolicyCleanupolicyConfiguration for cleanup.policy
uncleanLeaderElectionEnableBooleanConfiguration for unclean.leader.election.enable

BlackList

keytypedescription
valuesSet[String]A set of string that contains properties that cannot be changed
actionActionAction to take if the value is outside the specified range.

Integer

keytypedescription
minintMinimum value for the configuration.
maxintMaximum value for the configuration.
actionActionAction to take if the value is outside the specified range.
overrideValueintValue to override with (only applicable when action is set to OVERRIDE).

Long

keytypedescription
mindoubleMinimum value for the configuration.
maxdoubleMaximum value for the configuration.
actionActionAction to take if the value is outside the specified range.
overrideValuedoubleValue to override with (only applicable when action is set to OVERRIDE).

Cleanup Policy

keytypedescription
valuesSet[String]Value for the configuration, should be a set of string that contains values from delete, compact or specify both policies in a comma-separated list (eg: delete,compact).
actionActionAction to take if the value is outside the specified range.
overrideValueStringValue to override with (only applicable when action is set to OVERRIDE).

Boolean

keytypedescription
valueBooleanValue for the configuration. If action is OVERRIDE, will use this value for override value
actionActionAction to take if the value is outside the specified range.

Action

  • BLOCK → when fail, save in audit and return error.
  • INFO → execute API with wrong value, save in audit.
  • OVERRIDE → execute API with overrideValue (or value for others) values, save in audit the fact that we updated on the fly (with wrong value, and the one we used to fix them)

Example

{
"name": "myAlterTopicConfigPolicy",
"pluginClass": "io.conduktor.gateway.interceptor.safeguard.AlterTopicConfigPolicyPlugin",
"priority": 100,
"config": {
"topic": ".*",
"retentionMs": {
"min": 10,
"max": 100
},
"retentionBytes": {
"min": 10,
"max": 100,
"action": "BLOCK"
},
"segmentMs": {
"min": 10,
"max": 100,
"action": "INFO"
},
"segmentBytes": {
"min": 10,
"max": 100,
"action": "BLOCK"
},
"segmentJitterMs": {
"min": 10,
"max": 100,
"action": "INFO",
"overrideValue": 20
},
"flushMessages": {
"min": 10,
"max": 100,
"action": "OVERRIDE",
"overrideValue": 20
},
"flushMs": {
"min": 10,
"max": 100,
"action": "OVERRIDE",
"overrideValue": 20
},
"maxMessageBytes": {
"min": 10,
"max": 100,
"action": "OVERRIDE",
"overrideValue": 20
},
"minInsyncReplicas": {
"min": 10,
"max": 100,
"action": "OVERRIDE",
"overrideValue": 20
},
"cleanupPolicy": {
"value": [
"delete",
"compact"
],
"action": "OVERRIDE"
},
"uncleanLeaderElectionEnable": {
"value": false,
"action": "BLOCK"
}
}
}