Kafka AdminClient Tutorial: Managing Topics, Configurations, and Partitions with Java

Introduction

Typically administrators use the kafka-topics.sh script to manage Kafka topics, but many organizations prefer to embed topic‑management functions into internal systems. Since Kafka 0.11 the Java AdminClient has been available, and it continues to evolve.

Features

Topic management – create, delete, and query topics.

ACL management – configure and delete permissions.

Configuration management – set and query parameters for brokers, topics, users, client‑ids, etc.

Log directory management – modify and query replica log paths.

Partition management – add partitions to existing topics.

Message deletion – delete messages before a given offset.

Delegation Token management – create, update, expire, and query tokens.

Consumer group management – query, fetch offsets, and delete groups.

Preferred leader election – elect a specific broker as the leader for a partition.

Working Principle

The AdminClient uses a dual‑thread design: a front‑end thread that converts user operations into requests and pushes them to a queue, and a back‑end I/O thread ( kafka-admin-client-thread) that reads the queue, sends the requests to the appropriate broker, and stores the results for the front‑end thread to retrieve.

Usage

If you use Maven, add the following dependency:

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.6.5</version>
</dependency>

Building an AdminClient

/**
 * Create AdminClient instance using Properties
 */
public static AdminClient createAdminClientByProperties() {
    Properties prop = new Properties();
    // Kafka bootstrap server address
    prop.setProperty(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
    return AdminClient.create(prop);
}

/**
 * Create AdminClient instance using a Map
 */
public static AdminClient createAdminClientByMap() {
    Map<String, Object> conf = Maps.newHashMap();
    conf.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
    return AdminClient.create(conf);
}

Creating a Topic

private static final String TOPIC_NAME = "test_topic";

/**
 * Create a new topic
 */
public static void createTopic() {
    AdminClient adminClient = AdminSample.adminClient();
    Short replicationFactor = 1;
    NewTopic newTopic = new NewTopic(TOPIC_NAME, 1, replicationFactor);
    CreateTopicsResult result = adminClient.createTopics(Arrays.asList(newTopic));
    System.out.println("CreateTopicsResult : " + result);
    adminClient.close();
}

Listing Topics

/**
 * List all topics (including internal ones if requested)
 */
public static void topicList() throws Exception {
    AdminClient adminClient = adminClient();
    ListTopicsOptions options = new ListTopicsOptions();
    options.listInternal(true);
    ListTopicsResult result = adminClient.listTopics(options);
    Set<String> names = result.names().get();
    names.forEach(System.out::println);
    Collection<TopicListing> listings = result.listings().get();
    listings.forEach(t -> System.out.println(t.toString()));
    adminClient.close();
}

Deleting a Topic

/**
 * Delete a topic by name
 */
public static void delTopic() throws Exception {
    AdminClient adminClient = adminClient();
    DeleteTopicsResult deleteResult = adminClient.deleteTopics(Arrays.asList(TOPIC_NAME));
    deleteResult.all().get();
}

Describing a Topic

/**
 * Get description of one or more topics
 */
public static void describeTopics(List<String> topics) throws Exception {
    AdminClient adminClient = BuildAdminClient.createAdminClientByProperties();
    DescribeTopicsResult result = adminClient.describeTopics(topics);
    Map<String, TopicDescription> map = result.all().get();
    map.forEach((name, desc) -> System.out.printf("topic name = %s, desc = %s
", name, desc));
    adminClient.close();
}

Fetching Topic Configuration

Beyond broker settings, each topic has many configuration entries that can be retrieved via describeConfigs:

/**
 * Retrieve configuration of given topics
 */
public static void describeConfigTopics(List<String> topicNames) throws Exception {
    AdminClient adminClient = BuildAdminClient.createAdminClientByMap();
    List<ConfigResource> resources = Lists.newArrayListWithCapacity(64);
    topicNames.forEach(name -> resources.add(new ConfigResource(ConfigResource.Type.TOPIC, name)));
    DescribeConfigsResult result = adminClient.describeConfigs(resources);
    Map<ConfigResource, Config> configMap = result.all().get();
    configMap.forEach((res, cfg) -> System.out.printf("topic config %s = %s
", res, cfg));
    adminClient.close();
}

Increasing Topic Partitions

Partitions can only be increased. The following example shows how to add partitions to existing topics:

/**
 * Increase partition count for topics (cannot decrease)
 */
public static void updateTopicPartition(List<String> topicNames, Integer partitionNum) throws Exception {
    AdminClient adminClient = BuildAdminClient.createAdminClientByMap();
    Map<String, NewPartitions> newPartitions = Maps.newHashMap();
    topicNames.forEach(name -> newPartitions.put(name, NewPartitions.increaseTo(partitionNum)));
    CreatePartitionsResult result = adminClient.createPartitions(newPartitions);
    result.all().get();
    adminClient.close();
}

The AdminClient, introduced in Kafka 0.11, provides dozens of operational capabilities and continues to evolve; it is recommended to use it for all cluster management tasks instead of ZooKeeper‑based tools and to monitor its feature updates regularly.