KafkaIQ3.1 Dynamic Configuration & Runtime Changes
Module Overview
Goal: Teach safe runtime changes to Kafka configuration.
What You'll Learn:
- Static vs dynamic configurations
- Broker-level dynamic configs
- Topic-level configuration overrides
- Client quotas via dynamic configuration
- When changes require restarts
- Safe operational practices for config changes
What's NOT Covered Here:
- Performance tuning strategy (covered in Module 5.1)
- Monitoring metrics (covered in Module 4.x)
- Specific configuration values and tuning (covered throughout other modules)
Key Insight: Dynamic configuration enables zero-downtime changes for many settings, but understanding which configs are dynamic vs static is critical.
1. Static vs Dynamic Configuration
Configuration Types
STATIC CONFIGS:
├─ Set in server.properties file
├─ Require broker restart to take effect
├─ Example: broker.id, log.dirs, listeners
└─ Cannot be changed at runtime
DYNAMIC CONFIGS:
├─ Can be changed at runtime
├─ No broker restart needed
├─ Take effect immediately or after brief delay
└─ Stored in ZooKeeper or KRaft metadata
Dynamic vs Static Broker Configurations
Complete Reference Table:
┌──────────────────────────────┬──────────┬─────────────────────────────┐
│ Configuration │ Type │ Description │
├──────────────────────────────┼──────────┼─────────────────────────────┤
│ DYNAMIC (No Restart Required) │
├──────────────────────────────┼──────────┼─────────────────────────────┤
│ log.retention.ms │ Dynamic │ Time-based retention │
│ log.retention.bytes │ Dynamic │ Size-based retention │
│ log.segment.bytes │ Dynamic │ Segment file size │
│ log.cleaner.threads │ Dynamic │ Log compaction threads │
│ log.cleaner.min.compaction │ Dynamic │ Compaction dirty ratio │
│ .lag │ │ │
│ min.insync.replicas │ Dynamic │ Minimum ISR size │
│ max.connections │ Dynamic │ Max client connections │
│ max.connections.per.ip │ Dynamic │ Per-IP connection limit │
│ num.network.threads │ Dynamic* │ Network request threads │
│ num.io.threads │ Dynamic* │ Disk I/O threads │
│ background.threads │ Dynamic │ Background task threads │
│ compression.type │ Dynamic │ Broker-side compression │
│ leader.replication.throttled │ Dynamic │ Replication throttling │
│ .rate │ │ │
│ follower.replication │ Dynamic │ Follower fetch throttling │
│ .throttled.rate │ │ │
│ unclean.leader.election │ Dynamic │ Allow unclean elections │
│ .enable │ │ │
│ log.flush.interval.messages │ Dynamic │ Flush frequency │
│ log.flush.interval.ms │ Dynamic │ Flush timeout │
└──────────────────────────────┴──────────┴─────────────────────────────┘
┌──────────────────────────────┬──────────┬─────────────────────────────┐
│ STATIC (Restart Required) │
├──────────────────────────────┼──────────┼─────────────────────────────┤
│ broker.id │ Static │ Unique broker identifier │
│ log.dirs │ Static │ Data directory paths │
│ zookeeper.connect │ Static │ ZooKeeper connection string │
│ controller.quorum.voters │ Static │ KRaft controller quorum │
│ listeners │ Static │ Network listeners │
│ advertised.listeners │ Static │ Advertised endpoints │
│ inter.broker.listener.name │ Static │ Inter-broker communication │
│ security.inter.broker │ Static │ Inter-broker security │
│ .protocol │ │ │
│ num.partitions │ Static │ Default partition count │
│ default.replication.factor │ Static │ Default replication factor │
│ auto.create.topics.enable │ Static │ Auto-create topics │
│ delete.topic.enable │ Static │ Allow topic deletion │
│ offsets.topic.replication │ Static │ Offsets topic replication │
│ .factor │ │ │
│ transaction.state.log │ Static │ Transaction log replication │
│ .replication.factor │ │ │
└──────────────────────────────┴──────────┴─────────────────────────────┘
* Some configs like num.network.threads and num.io.threads may be dynamically
adjustable in recent Kafka versions, but changes are disruptive. Check your version.
How to Check if a Config is Dynamic:
# View all configs with their update modes
kafka-configs.sh --bootstrap-server localhost:9092 \
--entity-type brokers \
--entity-name 0 \
--describe --all
# Look for "updateMode" field:
# - READ_ONLY: Static (restart required)
# - PER_BROKER: Dynamic per-broker
# - CLUSTER_WIDE: Dynamic cluster-wide
Examples:
# DYNAMIC: Can be changed at runtime
kafka-configs.sh --bootstrap-server localhost:9092 \
--entity-type brokers --entity-name 0 --alter \
--add-config log.retention.ms=86400000
# Broker applies immediately, no restart needed ✓
# STATIC: Requires restart
# To change log.dirs, must edit server.properties:
log.dirs=/data1/kafka,/data2/kafka,/data3/kafka
# Then restart broker for change to take effect
systemctl restart kafka
How Dynamic Configs Work
LIFECYCLE OF DYNAMIC CONFIG CHANGE:
1. Admin issues config change
└─ via kafka-configs.sh or Admin API
2. Config stored in coordination system
└─ ZooKeeper: /config/... znodes
└─ KRaft: __cluster_metadata topic
3. Brokers watch for config changes
└─ Receive notification of update
4. Brokers apply new config
└─ Update in-memory configuration
5. New behavior takes effect
└─ Immediately for most configs
2. Configuration Scopes
Three Levels of Configuration
1. CLUSTER-DEFAULT (Broker-level defaults)
└─ Applies to all brokers unless overridden
2. PER-BROKER (Broker-specific)
└─ Applies to specific broker only
3. PER-TOPIC (Topic-level override)
└─ Applies to specific topic only
Precedence:
Per-Topic > Per-Broker > Cluster-Default > server.properties
Broker vs Topic Configuration Scope
Understanding which scope applies where:
┌────────────────┬────────────────────────────────┬─────────────────────────────┐
│ Scope │ Applies To │ Changed Via │
├────────────────┼────────────────────────────────┼─────────────────────────────┤
│ Broker-Level │ Global defaults for all topics │ server.properties OR │
│ (Cluster-Wide) │ Broker-specific settings │ kafka-configs.sh │
│ │ │ --entity-type brokers │
│ │ │ │
│ Examples: │ log.retention.hours=168 │ Affects all topics unless │
│ │ num.io.threads=8 │ overridden │
│ │ max.connections=1000 │ │
├────────────────┼────────────────────────────────┼─────────────────────────────┤
│ Topic-Level │ Per-topic overrides │ kafka-configs.sh │
│ │ Specific topic behavior │ --entity-type topics │
│ │ │ OR Admin API │
│ │ │ │
│ Examples: │ retention.ms=86400000 │ Overrides broker default │
│ │ cleanup.policy=compact │ for THIS topic only │
│ │ min.insync.replicas=2 │ │
└────────────────┴────────────────────────────────┴─────────────────────────────┘
PRECEDENCE RULE:
└─ Topic-level configuration ALWAYS overrides broker-level default
Real-World Example:
# Broker Default (server.properties)
log.retention.hours=168 # 7 days for all topics
# Result:
├─ Topic "orders": 7 days (uses broker default)
├─ Topic "analytics": 7 days (uses broker default)
└─ Topic "user-activity": 7 days (uses broker default)
# Now apply topic-level override
kafka-configs.sh --bootstrap-server localhost:9092 \
--entity-type topics \
--entity-name user-activity \
--alter \
--add-config retention.ms=86400000 # 1 day
# New Result:
├─ Topic "orders": 7 days (broker default)
├─ Topic "analytics": 7 days (broker default)
└─ Topic "user-activity": 1 day (OVERRIDE wins!)
# Topic-level config takes precedence ✓
When to Use Each Scope:
USE BROKER-LEVEL WHEN:
├─ Setting defaults for most/all topics
├─ Configuring broker behavior (threads, connections)
├─ Resource management (memory, CPU)
└─ Want simple, consistent configuration
USE TOPIC-LEVEL WHEN:
├─ Topic needs different retention than default
├─ Topic requires special cleanup policy (compaction)
├─ Topic has unique durability requirements
├─ Compliance requires topic-specific settings
└─ Testing configuration before cluster-wide change
Viewing Configuration Precedence:
# View topic config and its source
kafka-configs.sh --bootstrap-server localhost:9092 \
--entity-type topics \
--entity-name user-activity \
--describe --all
# Output shows source of each config:
# retention.ms=86400000 sensitive=false
# source=DYNAMIC_TOPIC_CONFIG ← Topic override
#
# compression.type=producer sensitive=false
# source=DEFAULT_CONFIG ← Using broker default
3. Viewing Configuration
View Broker Configuration
# View all configs for broker
kafka-configs.sh --bootstrap-server localhost:9092 \
--entity-type brokers \
--entity-name 0 \
--describe
# View cluster-wide defaults
kafka-configs.sh --bootstrap-server localhost:9092 \
--entity-type brokers \
--entity-default \
--describe
# Show only dynamic configs
kafka-configs.sh --bootstrap-server localhost:9092 \
--entity-type brokers \
--entity-name 0 \
--describe | grep DYNAMIC
View Topic Configuration
# View topic configs
kafka-configs.sh --bootstrap-server localhost:9092 \
--entity-type topics \
--entity-name orders \
--describe
# View all configs (including defaults)
kafka-configs.sh --bootstrap-server localhost:9092 \
--entity-type topics \
--entity-name orders \
--describe --all
View Client Quota Configuration
# View user quotas
kafka-configs.sh --bootstrap-server localhost:9092 \
--entity-type users \
--entity-name alice \
--describe
# View client quotas
kafka-configs.sh --bootstrap-server localhost:9092 \
--entity-type clients \
--entity-name my-producer \
--describe
4. Broker-Level Dynamic Configs
Common Dynamic Broker Configs
COMMONLY CHANGED DYNAMIC CONFIGS:
1. Logging Levels
└─ kafka.log.level
2. Replication Throttling
└─ leader.replication.throttled.rate
└─ follower.replication.throttled.rate
3. Connection Limits
└─ max.connections
└─ max.connections.per.ip
4. SSL/TLS Settings
└─ ssl.keystore.location
└─ ssl.truststore.location
5. Metrics Collection
└─ metric.reporters
6. Background Thread Counts
└─ num.recovery.threads.per.data.dir
└─ num.replica.fetchers
Changing Broker Config
# Set config for specific broker
kafka-configs.sh --bootstrap-server localhost:9092 \
--entity-type brokers \
--entity-name 0 \
--alter \
--add-config max.connections=2000
# Set cluster-wide default
kafka-configs.sh --bootstrap-server localhost:9092 \
--entity-type brokers \
--entity-default \
--alter \
--add-config max.connections=2000
# Remove config (revert to static value)
kafka-configs.sh --bootstrap-server localhost:9092 \
--entity-type brokers \
--entity-name 0 \
--alter \
--delete-config max.connections
Example: Change Log Level
# Increase log level for debugging
kafka-configs.sh --bootstrap-server localhost:9092 \
--entity-type brokers \
--entity-name 0 \
--alter \
--add-config kafka.server.KafkaApis=DEBUG
# Revert to default (INFO)
kafka-configs.sh --bootstrap-server localhost:9092 \
--entity-type brokers \
--entity-name 0 \
--alter \
--delete-config kafka.server.KafkaApis
5. Topic-Level Dynamic Configs
Common Dynamic Topic Configs
ALL TOPIC CONFIGS ARE DYNAMIC:
1. Retention
└─ retention.ms
└─ retention.bytes
2. Cleanup
└─ cleanup.policy
└─ min.cleanable.dirty.ratio
└─ segment.ms
3. Replication
└─ min.insync.replicas
└─ unclean.leader.election.enable
4. Compression
└─ compression.type
5. Segment
└─ segment.bytes
└─ segment.ms
6. Message Limits
└─ max.message.bytes
Changing Topic Config
# Change retention to 14 days
kafka-configs.sh --bootstrap-server localhost:9092 \
--entity-type topics \
--entity-name orders \
--alter \
--add-config retention.ms=1209600000
# Change multiple configs at once
kafka-configs.sh --bootstrap-server localhost:9092 \
--entity-type topics \
--entity-name orders \
--alter \
--add-config retention.ms=1209600000,compression.type=lz4,min.insync.replicas=2
# Remove config (revert to broker default)
kafka-configs.sh --bootstrap-server localhost:9092 \
--entity-type topics \
--entity-name orders \
--alter \
--delete-config retention.ms
6. Client Quotas (Dynamic Configuration)
Quota Types
CLIENT QUOTAS CONTROL:
1. Producer Quota
└─ Bytes produced per second
2. Consumer Quota
└─ Bytes consumed per second
3. Request Quota
└─ Request percentage (CPU time)
Applied to:
├─ User
├─ Client ID
└─ User + Client ID combination
Setting Client Quotas
# Set producer quota for user
kafka-configs.sh --bootstrap-server localhost:9092 \
--entity-type users \
--entity-name alice \
--alter \
--add-config producer_byte_rate=1048576,consumer_byte_rate=2097152
# Set quota for client ID
kafka-configs.sh --bootstrap-server localhost:9092 \
--entity-type clients \
--entity-name my-producer \
--alter \
--add-config producer_byte_rate=10485760
# Set quota for user + client combination
kafka-configs.sh --bootstrap-server localhost:9092 \
--entity-type users \
--entity-name alice \
--entity-type clients \
--entity-name my-app \
--alter \
--add-config producer_byte_rate=2097152
# Set default user quota
kafka-configs.sh --bootstrap-server localhost:9092 \
--entity-type users \
--entity-default \
--alter \
--add-config producer_byte_rate=1048576,consumer_byte_rate=2097152
More details in Module 6.1: Quotas & Resource Management.
7. When Restarts Are Required
Configs Requiring Restart
STATIC CONFIGS (MUST RESTART):
BROKER IDENTITY:
├─ broker.id
├─ broker.rack
└─ reserved.broker.max.id
NETWORKING:
├─ listeners
├─ advertised.listeners
└─ listener.security.protocol.map
STORAGE:
├─ log.dirs
└─ num.io.threads
COORDINATION:
├─ zookeeper.connect (ZK mode)
└─ controller.quorum.voters (KRaft mode)
REPLICATION:
├─ num.replica.fetchers
└─ replica.fetch.max.bytes
THREADING:
└─ num.network.threads
Determining If Restart Needed
RULES OF THUMB:
NO RESTART NEEDED:
├─ Topic-level configs (always dynamic)
├─ Client quotas (always dynamic)
├─ Logging levels (always dynamic)
└─ Throttling limits (always dynamic)
RESTART NEEDED:
├─ Network configuration (listeners, ports)
├─ Storage paths (log.dirs)
├─ Thread pool sizes (num.network.threads)
├─ Broker identity (broker.id, rack)
└─ Coordination settings (ZK connect, KRaft voters)
WHEN IN DOUBT:
└─ Check Kafka documentation for specific config
└─ Test in non-production first
8. Safe Configuration Change Practices
Pre-Change Checklist
BEFORE CHANGING CONFIG:
☑ Document current value
☑ Understand impact of change
☑ Test in non-production first
☑ Have rollback plan
☑ Check if restart required
☑ Schedule maintenance window (if restart needed)
☑ Notify stakeholders
☑ Monitor metrics before change (baseline)
Change Process
STEP-BY-STEP:
1. Verify current config
└─ kafka-configs.sh --describe
2. Apply change
└─ kafka-configs.sh --alter --add-config
3. Verify change applied
└─ kafka-configs.sh --describe (check new value)
4. Monitor impact
├─ Check metrics for expected changes
├─ Watch for errors in logs
└─ Validate client behavior
5. Rollback if needed
└─ kafka-configs.sh --alter --delete-config
6. Document change
└─ Update runbooks, wiki, change log
Rollback Strategy
IF CHANGE CAUSES ISSUES:
1. Immediate rollback
└─ Delete config to revert to previous value
└─ Or set explicitly to old value
2. Monitor recovery
└─ Watch metrics return to normal
3. Investigate root cause
└─ Why did change cause issues?
4. Plan alternative approach
└─ Different config value
└─ Phased rollout
9. Common Dynamic Configuration Scenarios
Scenario 1: Increase Retention for Topic
GOAL: Extend data retention from 7 to 14 days
STEPS:
1. Check current retention
kafka-topics.sh --describe --topic orders
2. Change retention
kafka-configs.sh --bootstrap-server localhost:9092 \
--entity-type topics \
--entity-name orders \
--alter \
--add-config retention.ms=1209600000
3. Verify
kafka-topics.sh --describe --topic orders
IMPACT:
├─ Immediate: No impact (data not deleted yet)
├─ Over time: Disk usage increases
└─ Monitoring: Watch disk usage trends
Scenario 2: Throttle Replication During Reassignment
GOAL: Limit replication bandwidth to 50 MB/s during partition reassignment
STEPS:
1. Set throttle on all brokers
kafka-configs.sh --bootstrap-server localhost:9092 \
--entity-type brokers \
--entity-name 0 \
--alter \
--add-config leader.replication.throttled.rate=52428800,follower.replication.throttled.rate=52428800
# Repeat for all brokers
2. Execute reassignment
kafka-reassign-partitions.sh --execute
3. Monitor progress
kafka-reassign-partitions.sh --verify
4. Remove throttle after completion
kafka-configs.sh --bootstrap-server localhost:9092 \
--entity-type brokers \
--entity-name 0 \
--alter \
--delete-config leader.replication.throttled.rate,follower.replication.throttled.rate
IMPACT:
├─ Slower reassignment
├─ Less impact on production traffic
└─ Better control over resource usage
Scenario 3: Enable Debug Logging
GOAL: Debug connectivity issues for specific broker
STEPS:
1. Enable DEBUG logging
kafka-configs.sh --bootstrap-server localhost:9092 \
--entity-type brokers \
--entity-name 0 \
--alter \
--add-config kafka.network.SocketServer=DEBUG
2. Reproduce issue while capturing logs
3. Revert to INFO level
kafka-configs.sh --bootstrap-server localhost:9092 \
--entity-type brokers \
--entity-name 0 \
--alter \
--delete-config kafka.network.SocketServer
IMPACT:
├─ Increased log volume
├─ Potential performance impact
└─ Should be temporary (minutes, not hours)
Scenario 4: Change min.insync.replicas
GOAL: Increase durability guarantee from 1 to 2
STEPS:
1. Check current ISR size
kafka-topics.sh --describe --topic orders
2. Change min.insync.replicas
kafka-configs.sh --bootstrap-server localhost:9092 \
--entity-type topics \
--entity-name orders \
--alter \
--add-config min.insync.replicas=2
3. Verify
kafka-topics.sh --describe --topic orders
IMPACT:
├─ Higher durability (leader + 1 follower must ack)
├─ Lower availability (writes fail if ISR < 2)
├─ Producers using acks=all will experience change
└─ Ensure replication factor ≥ 2!
10. Bulk Configuration Changes
Applying Configs to Multiple Topics
# Get all topics
kafka-topics.sh --bootstrap-server localhost:9092 --list > topics.txt
# Apply config to each topic
while read topic; do
kafka-configs.sh --bootstrap-server localhost:9092 \
--entity-type topics \
--entity-name "$topic" \
--alter \
--add-config retention.ms=604800000
done < topics.txt
Using Java Admin API (Programmatic Configuration)
For automation and integration, use the Admin API to manage configurations programmatically.
Setup Admin Client
import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.config.ConfigResource;
import java.util.*;
import java.util.concurrent.ExecutionException;
public class DynamicConfigManager {
private AdminClient adminClient;
public void init(String bootstrapServers) {
Properties props = new Properties();
props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
props.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, 30000);
adminClient = AdminClient.create(props);
}
public void close() {
adminClient.close();
}
}
Example 1: Alter Topic Configuration
public void setTopicRetention(String topicName, long retentionMs)
throws ExecutionException, InterruptedException {
// Define the topic resource
ConfigResource resource = new ConfigResource(
ConfigResource.Type.TOPIC,
topicName
);
// Create config entry
ConfigEntry retentionConfig = new ConfigEntry(
"retention.ms",
String.valueOf(retentionMs)
);
// Create alter operation
AlterConfigOp op = new AlterConfigOp(
retentionConfig,
AlterConfigOp.OpType.SET
);
// Apply the configuration
Map<ConfigResource, Collection<AlterConfigOp>> configs =
Collections.singletonMap(resource, Collections.singleton(op));
adminClient.incrementalAlterConfigs(configs).all().get();
System.out.println("✓ Updated retention for topic: " + topicName);
}
// Usage
configManager.setTopicRetention("orders", 86400000L); // 1 day
Example 2: View Topic Configuration
public void describeTopicConfig(String topicName)
throws ExecutionException, InterruptedException {
ConfigResource resource = new ConfigResource(
ConfigResource.Type.TOPIC,
topicName
);
DescribeConfigsResult result = adminClient.describeConfigs(
Collections.singleton(resource)
);
Map<ConfigResource, Config> configs = result.all().get();
Config topicConfig = configs.get(resource);
System.out.println("Configuration for topic: " + topicName);
System.out.println("─".repeat(50));
for (ConfigEntry entry : topicConfig.entries()) {
System.out.printf("%-30s = %s%n", entry.name(), entry.value());
// Show if it's a dynamic override or default
if (entry.source() == ConfigEntry.ConfigSource.DYNAMIC_TOPIC_CONFIG) {
System.out.println(" (dynamic override)");
}
}
}
Example 3: Delete Config Override (Revert to Default)
public void deleteTopicConfigOverride(String topicName, String configKey)
throws ExecutionException, InterruptedException {
ConfigResource resource = new ConfigResource(
ConfigResource.Type.TOPIC,
topicName
);
// Create DELETE operation
AlterConfigOp deleteOp = new AlterConfigOp(
new ConfigEntry(configKey, null), // value=null for delete
AlterConfigOp.OpType.DELETE
);
Map<ConfigResource, Collection<AlterConfigOp>> configs =
Collections.singletonMap(resource, Collections.singleton(deleteOp));
adminClient.incrementalAlterConfigs(configs).all().get();
System.out.println("✓ Deleted override for: " + configKey);
System.out.println(" Topic will now use broker default");
}
// Usage
configManager.deleteTopicConfigOverride("orders", "retention.ms");
Example 4: Set Multiple Configs at Once
public void configureHighThroughputTopic(String topicName)
throws ExecutionException, InterruptedException {
ConfigResource resource = new ConfigResource(
ConfigResource.Type.TOPIC,
topicName
);
// Define multiple configurations
List<AlterConfigOp> ops = Arrays.asList(
new AlterConfigOp(
new ConfigEntry("compression.type", "lz4"),
AlterConfigOp.OpType.SET
),
new AlterConfigOp(
new ConfigEntry("min.insync.replicas", "2"),
AlterConfigOp.OpType.SET
),
new AlterConfigOp(
new ConfigEntry("retention.ms", "604800000"), // 7 days
AlterConfigOp.OpType.SET
),
new AlterConfigOp(
new ConfigEntry("segment.ms", "3600000"), // 1 hour
AlterConfigOp.OpType.SET
)
);
Map<ConfigResource, Collection<AlterConfigOp>> configs =
Collections.singletonMap(resource, ops);
adminClient.incrementalAlterConfigs(configs).all().get();
System.out.println("✓ Configured topic for high throughput: " + topicName);
}
Example 5: Bulk Config Update (Pattern Matching)
public void setBulkRetention(String topicPrefix, long retentionMs)
throws ExecutionException, InterruptedException {
// Get all topics matching pattern
Set<String> allTopics = adminClient.listTopics().names().get();
Set<String> matchingTopics = allTopics.stream()
.filter(t -> t.startsWith(topicPrefix))
.collect(Collectors.toSet());
System.out.println("Found " + matchingTopics.size() + " topics matching: " + topicPrefix);
// Prepare configs for all matching topics
Map<ConfigResource, Collection<AlterConfigOp>> configs = new HashMap<>();
for (String topic : matchingTopics) {
ConfigResource resource = new ConfigResource(
ConfigResource.Type.TOPIC,
topic
);
AlterConfigOp op = new AlterConfigOp(
new ConfigEntry("retention.ms", String.valueOf(retentionMs)),
AlterConfigOp.OpType.SET
);
configs.put(resource, Collections.singleton(op));
}
// Apply all configs
adminClient.incrementalAlterConfigs(configs).all().get();
System.out.println("✓ Updated retention for " + matchingTopics.size() + " topics");
}
// Usage
configManager.setBulkRetention("app-", 86400000L); // All topics starting with "app-"
Example 6: Set Broker-Level Dynamic Config
public void setBrokerConfig(int brokerId, String configKey, String configValue)
throws ExecutionException, InterruptedException {
ConfigResource resource = new ConfigResource(
ConfigResource.Type.BROKER,
String.valueOf(brokerId)
);
AlterConfigOp op = new AlterConfigOp(
new ConfigEntry(configKey, configValue),
AlterConfigOp.OpType.SET
);
Map<ConfigResource, Collection<AlterConfigOp>> configs =
Collections.singletonMap(resource, Collections.singleton(op));
adminClient.incrementalAlterConfigs(configs).all().get();
System.out.println("✓ Updated broker " + brokerId + " config: " + configKey);
}
// Usage
configManager.setBrokerConfig(0, "log.retention.hours", "168"); // 7 days
Complete Example: Config Manager Class
import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.config.ConfigResource;
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;
public class KafkaConfigManager implements AutoCloseable {
private final AdminClient adminClient;
public KafkaConfigManager(String bootstrapServers) {
Properties props = new Properties();
props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
props.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, 30000);
this.adminClient = AdminClient.create(props);
}
public void alterTopicConfig(String topic, String key, String value)
throws ExecutionException, InterruptedException {
ConfigResource resource = new ConfigResource(ConfigResource.Type.TOPIC, topic);
AlterConfigOp op = new AlterConfigOp(
new ConfigEntry(key, value),
AlterConfigOp.OpType.SET
);
Map<ConfigResource, Collection<AlterConfigOp>> configs =
Collections.singletonMap(resource, Collections.singleton(op));
adminClient.incrementalAlterConfigs(configs).all().get();
}
public void describeTopicConfig(String topic)
throws ExecutionException, InterruptedException {
ConfigResource resource = new ConfigResource(ConfigResource.Type.TOPIC, topic);
DescribeConfigsResult result = adminClient.describeConfigs(
Collections.singleton(resource)
);
Config config = result.all().get().get(resource);
config.entries().forEach(entry ->
System.out.println(entry.name() + " = " + entry.value())
);
}
public void deleteTopicConfigOverride(String topic, String key)
throws ExecutionException, InterruptedException {
ConfigResource resource = new ConfigResource(ConfigResource.Type.TOPIC, topic);
AlterConfigOp op = new AlterConfigOp(
new ConfigEntry(key, null),
AlterConfigOp.OpType.DELETE
);
Map<ConfigResource, Collection<AlterConfigOp>> configs =
Collections.singletonMap(resource, Collections.singleton(op));
adminClient.incrementalAlterConfigs(configs).all().get();
}
@Override
public void close() {
adminClient.close();
}
// Main method for testing
public static void main(String[] args) {
try (KafkaConfigManager manager = new KafkaConfigManager("localhost:9092")) {
// Set retention to 1 day
manager.alterTopicConfig("orders", "retention.ms", "86400000");
System.out.println("✓ Updated retention");
// View configuration
manager.describeTopicConfig("orders");
// Revert to default
manager.deleteTopicConfigOverride("orders", "retention.ms");
System.out.println("✓ Reverted to default retention");
} catch (Exception e) {
e.printStackTrace();
}
}
}
11. Configuration Best Practices
DO
✅ Test changes in non-production first
✅ Document all configuration changes
✅ Monitor metrics before and after
✅ Use version control for static configs
✅ Have rollback plan ready
✅ Change one config at a time (for troubleshooting)
✅ Use cluster-wide defaults when appropriate
✅ Remove dynamic overrides when no longer needed
DON'T
❌ Change multiple configs simultaneously
❌ Apply untested configs in production
❌ Forget to monitor after changes
❌ Leave debug logging enabled permanently
❌ Use dynamic configs for values that should be static
❌ Ignore warnings in logs after config changes
❌ Change configs during peak traffic (unless necessary)
❌ Apply configs blindly without understanding impact
12. Configuration Management Tools
Infrastructure as Code
RECOMMENDED APPROACHES:
1. Ansible
└─ Manage server.properties via templates
└─ Apply dynamic configs via kafka-configs.sh tasks
2. Terraform with Kafka Provider
└─ Manage topics and configs declaratively
└─ Track changes in version control
3. GitOps
└─ Store configs in Git
└─ CI/CD pipeline applies changes
└─ Audit trail built-in
4. Configuration Management Platform
└─ Confluent Control Center
└─ LinkedIn Cruise Control
└─ Custom tooling
13. Key Takeaways
You've learned:
✅ Dynamic configs can be changed at runtime (no restart)
✅ Static configs require broker restart
✅ Three config scopes: cluster-default, per-broker, per-topic
✅ kafka-configs.sh is the primary tool for dynamic changes
✅ Topic configs are always dynamic
✅ Client quotas can be set dynamically
✅ Rollback is easy (delete config or set to old value)
✅ Test changes in non-production first
✅ Monitor impact of every configuration change
14. Configuration Change Checklist
PRE-CHANGE:
☑ Document current value
☑ Understand impact
☑ Test in non-production
☑ Have rollback plan
☑ Check if restart needed
☑ Schedule window (if restart)
DURING CHANGE:
☑ Apply change
☑ Verify applied
☑ Monitor logs
☑ Check metrics
POST-CHANGE:
☑ Validate expected behavior
☑ Monitor for period (15-60 min)
☑ Document change
☑ Update runbooks
Summary
Safe configuration management:
Dynamic Configuration =
├─ Zero-downtime changes
├─ Immediate effect
├─ Easy rollback
├─ Powerful tool for operations
└─ Requires careful planning
Safe Practice =
├─ Test first
├─ Monitor always
├─ Rollback if needed
├─ Document everything
└─ Change deliberately
Remember: Dynamic configuration is a powerful operational tool, but power requires responsibility. Always test, always monitor, always have a rollback plan.
Next up: Module 4.1 — Core Kafka Metrics — Build a shared vocabulary for observability.