Bulk Topic Management with Kafka Admin Client API
Manual topic creation becomes a bottleneck once you run Apache Kafka at any real scale. Whether you are onboarding a microservice that needs a dozen topics or enforcing governance across a multi-tenant cluster, driving everything through kafka-topics.sh is slow, error-prone, and awkward to embed in CI/CD. Each invocation of the shell tool spins up a fresh JVM and a throwaway AdminClient under the hood. The Java and Python Admin Client APIs let you do the same work programmatically: keep one connection open, batch operations, and enforce policy in code. This guide covers a production approach to automating topic lifecycle operations.
Why Bulk Topic Management Matters at Scale
In a production Kafka deployment, topics are provisioned, reconfigured, and decommissioned in lockstep with application deployments. A single event-driven service might need topics for commands, events, snapshots, and a dead-letter queue, each with its own partition count, replication factor, and cleanup policy. Doing this by hand invites drift: a missed min.insync.replicas, a topic created with replication factor 1, or a compacted topic that was meant to be time-retained.
The Admin Client addresses this. The shell script kafka-topics.sh constructs an AdminClient, runs one request, and tears it down. A long-lived client instead keeps a persistent connection to the cluster and reuses it across many createTopics, describeConfigs, incrementalAlterConfigs, and createPartitions calls. Each of those requests returns a future, so you can fan out many operations concurrently and join on the results. More importantly, it gives you one place to enforce policy: validate names against a regex, require a minimum replication factor, or reject topics that lack required metadata. This turns governance from a documentation exercise into an executable contract, and it complements the broader discipline of Topics, Partitions & Data Modeling and Topic Naming Conventions and Governance Best Practices.
Setting Up the Admin Client for Production
The Admin Client manages its own network threads and connection pool, so treat it as a singleton for the lifetime of your tool rather than constructing one per operation. Configure it to mirror the security and timeout requirements of the target cluster.
List several bootstrap servers rather than one. The client only needs to reach a single listed broker to fetch full cluster metadata, but supplying several improves the odds of a successful initial connection during a rolling restart.
import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.AdminClientConfig;
import java.util.Properties;
public class AdminClientFactory {
public static Admin createAdminClient(String clusterName) {
Properties props = new Properties();
props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,
"broker1:9092,broker2:9092,broker3:9092");
props.put(AdminClientConfig.CLIENT_ID_CONFIG,
"topic-manager-" + clusterName);
// Per-RPC timeout; an operation may attempt several RPCs.
props.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, 30000);
// Upper bound for a whole operation, retries included.
props.put(AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, 60000);
// Security: set these only if your cluster requires SASL/SSL.
props.put("security.protocol", "SASL_SSL");
props.put("sasl.mechanism", "PLAIN");
return Admin.create(props);
}
}Admin.create(Properties) returns an Admin instance (AdminClient, the older abstract class, is still supported and AdminClient.create remains valid). The client.id is not cosmetic: it appears in broker request logs and client JMX metrics, which makes it the fastest way to attribute traffic during an incident. Prefer default.api.timeout.ms to bound an entire operation; request.timeout.ms only bounds a single underlying RPC. AdminClientConfig.RETRIES_CONFIG exists, but for the Admin Client the timeout settings are the more reliable lever, since the client retries retriable errors until the API timeout elapses. Always close() the client on shutdown, ideally with try-with-resources or a shutdown hook.
For Python tooling, the confluent-kafka library (a binding over librdkafka) provides an equivalent AdminClient. Install it with pip install confluent-kafka:
from confluent_kafka.admin import AdminClient
admin_client = AdminClient({
'bootstrap.servers': 'broker1:9092,broker2:9092,broker3:9092',
'client.id': 'topic-manager-prod',
'request.timeout.ms': 30000,
'security.protocol': 'SASL_SSL',
'sasl.mechanisms': 'PLAIN',
})Note the property-name differences from Java: it is sasl.mechanisms (plural) here, and there is no retries key. Retries are handled internally by librdkafka; use request.timeout.ms, socket.timeout.ms, and broker-side settings to influence behavior.
Batch Topic Creation with Policy Enforcement
createTopics accepts a collection of NewTopic objects and returns a CreateTopicsResult. Call result.all() to obtain a KafkaFuture<Void> that completes once every topic is created and fails if any single creation fails, or call result.values() to get a per-topic Map<String, KafkaFuture<Void>> so you can handle partial failures individually. The brokers process each topic creation independently; the call is not transactional, so some topics may be created while others fail.
The function below validates names, requires a minimum replication factor (rejecting rather than silently raising it, so a misconfigured manifest fails loudly), and applies default configs before submitting.
import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.common.config.TopicConfig;
import java.util.*;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
public class BulkTopicCreator {
private static final Pattern NAME_PATTERN =
Pattern.compile("^[a-z][a-z0-9-]*\\.[a-z][a-z0-9-]*\\.[a-z][a-z0-9-]*$");
private static final int MIN_REPLICATION = 3;
public static void createTopics(Admin client, List<TopicSpec> specs)
throws Exception {
List<NewTopic> newTopics = specs.stream().map(spec -> {
if (!NAME_PATTERN.matcher(spec.name).matches()) {
throw new IllegalArgumentException(
"Invalid topic name: " + spec.name);
}
if (spec.replicationFactor < MIN_REPLICATION) {
throw new IllegalArgumentException(
"Replication factor for " + spec.name +
" must be >= " + MIN_REPLICATION);
}
Map<String, String> configs = new HashMap<>();
configs.put(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, "2");
configs.put(TopicConfig.CLEANUP_POLICY_CONFIG,
TopicConfig.CLEANUP_POLICY_DELETE);
configs.putAll(spec.configs); // per-topic overrides win
return new NewTopic(spec.name, spec.partitions, spec.replicationFactor)
.configs(configs);
}).collect(Collectors.toList());
client.createTopics(newTopics).all().get(); // throws on any failure
}
}TopicSpec here is a plain value object (name, partition count, replication factor, and a Map<String, String> of configs) that you would hydrate from a YAML or JSON manifest. With min.insync.replicas=2 against a replication factor of 3, the topic stays writable while one replica is offline but rejects writes if two are down, which is the standard durability posture. The constructor used is NewTopic(String name, int numPartitions, short replicationFactor); if you supply explicit replica assignments instead, use the Map<Integer, List<Integer>> constructor and leave the count/factor to the broker.
Idempotent Configuration Reconciliation
Topic configs frequently need to change after creation: extending retention, switching cleanup policy, or tightening min.insync.replicas. The legacy alterConfigs API replaces the entire config set for a resource, which is dangerous because it silently clears any key you forget to include. Use incrementalAlterConfigs (Java) / incremental_alter_configs (Python) instead, which only touches the keys you name. This API was added in KIP-339 and requires brokers running Apache Kafka 2.3.0 or later. Updates are not transactional across resources, but the configs for a single resource are applied atomically.
To make reconciliation idempotent, read the current config with describe_configs, compare against the desired state, and only submit SET operations for keys whose values actually differ. Re-running against an already-converged cluster then submits nothing.
from confluent_kafka.admin import (
AdminClient, ConfigResource, ConfigEntry, AlterConfigOpType,
)
def reconcile_topic_configs(admin_client, desired_configs):
"""desired_configs: {topic_name: {config_key: value}}"""
resources = [
ConfigResource(ConfigResource.Type.TOPIC, topic)
for topic in desired_configs
]
describe_futures = admin_client.describe_configs(resources)
alter_resources = []
for resource, future in describe_futures.items():
topic = resource.name
current = {name: entry.value
for name, entry in future.result().items()}
desired = desired_configs[topic]
alterations = [] # scoped per topic
for key, value in desired.items():
if current.get(key) != str(value):
alterations.append(
ConfigEntry(key, str(value),
incremental_operation=AlterConfigOpType.SET)
)
if alterations:
alter_resources.append(
ConfigResource(ConfigResource.Type.TOPIC, topic,
incremental_configs=alterations)
)
if alter_resources:
alter_futures = admin_client.incremental_alter_configs(alter_resources)
for future in alter_futures.values():
future.result() # raises on failuredescribe_configs returns a dict of futures keyed by the ConfigResource you passed in; each future resolves to a dict of {config_name: ConfigEntry}, and ConfigEntry exposes .name and .value (values are always strings, hence the str(value) comparison). The alterations list is rebuilt inside the per-topic loop so configs from one topic never leak into another. This loop is safe to run repeatedly, for example from a GitOps reconciliation job. In modern KRaft-based clusters the controller applies these changes; there is no ZooKeeper involved. For the full set of topic config keys and their semantics, see the Apache Kafka topic configuration reference.
Partition Expansion at Scale
As throughput grows you will increase the partition count of existing topics. createPartitions takes a Map<String, NewPartitions> and returns a CreatePartitionsResult. Note two hard constraints: you can only ever increase the count (increaseTo rejects a value that is not strictly greater than the current count), and increasing partitions on a keyed topic breaks the key-to-partition mapping for existing keys, so plan this carefully for compacted or ordering-sensitive topics.
import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.NewPartitions;
import java.util.HashMap;
import java.util.Map;
public class BulkPartitionExpander {
private static final int MAX_PARTITIONS_PER_BROKER = 4000;
public static void expandPartitions(Admin adminClient,
Map<String, Integer> topicNewCounts,
int brokerCount) throws Exception {
Map<String, NewPartitions> expansions = new HashMap<>();
for (var entry : topicNewCounts.entrySet()) {
int newCount = entry.getValue();
// Ceiling division: worst-case partitions on a single broker.
int perBroker = (newCount + brokerCount - 1) / brokerCount;
if (perBroker > MAX_PARTITIONS_PER_BROKER) {
System.err.printf(
"Warning: %s reaches ~%d partitions/broker (limit %d)%n",
entry.getKey(), perBroker, MAX_PARTITIONS_PER_BROKER);
}
expansions.put(entry.getKey(), NewPartitions.increaseTo(newCount));
}
adminClient.createPartitions(expansions).all().get();
}
}The per-broker guard uses ceiling division so it does not underestimate the load on the busiest broker. The 4000 figure is a conservative working ceiling; the practical limit depends on heap, file descriptors, and recovery time, not a fixed number.
Adding partitions does not rebalance existing data across brokers. To move replicas, the Admin Client exposes alterPartitionReassignments, which submits a reassignment plan (and accepts Optional.empty() for a partition to cancel an in-flight reassignment). Building safe, throttled plans for a whole cluster is non-trivial, so most teams run Cruise Control rather than hand-writing plans. Treat alterPartitionReassignments as the building block Cruise Control and similar tools use, not as something you typically drive directly in bulk.
Integrating Bulk Operations into CI/CD Pipelines
The pattern that ties this together is a topic manifest checked into the application repository, reconciled by the pipeline before the application deploys. The manifest becomes the source of truth; the pipeline becomes the enforcement point.
topics:
- name: orders.commands.v1
partitions: 12
replicationFactor: 3
configs:
cleanup.policy: delete
retention.ms: 259200000 # 3 days
- name: orders.events.v1
partitions: 24
replicationFactor: 3
configs:
cleanup.policy: compact
min.compaction.lag.ms: 3600000
- name: orders.snapshots.v1
partitions: 6
replicationFactor: 3
configs:
cleanup.policy: "compact,delete" # compact, then age out old segmentscleanup.policy: compact,delete is a valid combined policy (KIP-71): segments are compacted and also aged out by retention.ms/retention.bytes. Your tooling (Jenkins, GitHub Actions, or a custom operator) parses the manifest, validates it against a JSON Schema encoding your governance rules, then calls the Admin Client to create new topics and reconcile configs on existing ones. This is the executable form of the governance described in Topic Naming Conventions and Governance Best Practices.
Error Handling and Observability
Bulk operations fail partially. A createTopics call can succeed for nine topics and fail for one on a quota or policy violation. Distinguish transient errors such as NotControllerException (the controller moved; retry) from permanent ones such as PolicyViolationException (a broker create.topic.policy rejected the request; do not retry). TopicExistsException is usually benign in a reconcile loop and can be treated as success.
import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.common.errors.NotControllerException;
import org.apache.kafka.common.errors.PolicyViolationException;
import org.apache.kafka.common.errors.TopicExistsException;
import java.util.List;
import java.util.concurrent.ExecutionException;
public class AdminClientRetryHandler {
public static void handleCreateTopics(Admin client,
List<NewTopic> topics,
int maxRetries) throws Exception {
for (int attempt = 0; attempt < maxRetries; attempt++) {
try {
client.createTopics(topics).all().get();
return;
} catch (ExecutionException e) {
Throwable cause = e.getCause();
if (cause instanceof TopicExistsException) {
System.out.println("Topics already exist, treating as done");
return;
} else if (cause instanceof NotControllerException) {
System.err.println("Controller moved, retrying...");
Thread.sleep(500L * (attempt + 1)); // simple backoff
} else if (cause instanceof PolicyViolationException) {
throw new RuntimeException(
"Permanent policy violation: " + cause.getMessage(), cause);
} else {
throw new RuntimeException("Unexpected error", cause);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new RuntimeException("Interrupted", e);
}
}
throw new RuntimeException("Exhausted retries creating topics");
}
}Note that all().get() throws on the first failure, so for true per-topic handling iterate the values() map and inspect each future independently. The Admin Client also publishes client-side metrics over JMX (request latency, in-flight requests, error rates) under the kafka.admin.client MBean domain. Instrument your wrapper as well — count of topics created, operation latency, and failure rate — so you can alert when reconciliation stalls, which is often the first visible symptom of a controller or quota problem.
Conclusion
Bulk topic management with the Kafka Admin Client API replaces brittle shell scripts with programmatic, policy-enforcing workflows. A long-lived singleton client, batch creation with up-front validation, idempotent config reconciliation via incrementalAlterConfigs, careful partition expansion, and manifest-driven CI/CD together eliminate configuration drift and make governance executable. Each piece maps to a real, version-checked API, so you can adopt them incrementally and extend toward self-service provisioning as your platform matures.