Using Admin API

1. Creating Admin Client

Property	Description	Default Value	Notes
bootstrap.servers	Comma-separated list of broker addresses	localhost:9092	Required for client connection
client.id	Unique identifier for the client	admin-client-1	Used for logging and monitoring
request.timeout.ms	Timeout for admin requests	30000	Time in milliseconds
retries	Number of retries for failed requests	0	Set to 0 for no retries

Example: Creating Admin Client

const { AdminClient } = require('kafkajs');

const admin = AdminClient.create({
  brokers: ['localhost:9092'],
  clientId: 'my-admin-client',
  requestTimeout: 30000,
});

module.exports = admin;

2. Creating Topics

Example: Create a topic with default configuration

const { AdminClient } = require('kafkajs');

async function createTopic() {
  const admin = AdminClient.create({
    brokers: ['localhost:9092'],
    clientId: 'topic-creator'
  });

  await admin.createTopics({
    topics: [{
      topic: 'my-new-topic',
      numPartitions: 1,
      replicationFactor: 1
    }]
  });

  console.log('Topic created successfully');
  await admin.disconnect();
}

createTopic().catch(console.error);

Property	Description	Default Value
topic	Name of the topic to create	None
numPartitions	Number of partitions for the topic	1
replicationFactor	Number of replicas for each partition	1
configEntries	Topic-level configuration properties	None

Note: The replication factor must be less than or equal to the number of brokers in the cluster.

Warning: Creating topics with a replication factor greater than the number of brokers will result in an error.

3. Listing Topics

Example: List all topics

const { AdminClient } = require(&#x27;kafkajs&#x27;);

const admin = AdminClient.create({
  brokers: [&#x27;localhost:9092&#x27;],
  clientId: &#x27;admin-client&#x27;
});

async function listTopics() {
  try {
    const topics = await admin.listTopics();
    console.log(&#x27;Topics:&#x27;, topics);
  } catch (err) {
    console.error(&#x27;Failed to list topics:&#x27;, err);
  }
}

listTopics();

Property	Description	Default
topics	List of topic names	None

Note: The listTopics() method returns a list of topic names as strings.

4. Describing Topics

Property	Description	Example
topic	Name of the Kafka topic	`user-events`
partitions	Number of partitions in the topic	`6`
replication.factor	Replication factor for the topic	`3`
retention.ms	Time in milliseconds to retain messages	`604800000`
cleanup.policy	Policy for cleaning up old messages (delete or compact)	`delete`

Example: Describe a Topic

kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic user-events

Example: Describe Topics with Detailed Information

kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic user-events --verbose

5. Deleting Topics

Property	Description	Default Value	Notes
topic.name	Name of the topic to delete	None	Required parameter
admin.delete.topics.timeout.ms	Timeout for topic deletion operation	120000	Milliseconds

Example: Delete a Kafka Topic

const { AdminClient } = require('kafkajs');

async function deleteTopic() {
  const admin = AdminClient.create({
    brokers: ['localhost:9092'],
    clientId: 'admin-client'
  });

  try {
    await admin.deleteTopics({
      topics: ['my-topic']
    });
    console.log('Topic deleted successfully');
  } catch (error) {
    console.error('Error deleting topic:', error);
  } finally {
    await admin.disconnect();
  }
}

deleteTopic();

Note: Deleting a topic is irreversible and will remove all data associated with it.

Warning: Ensure that no consumers or producers are actively using the topic before deletion.

6. Creating Partitions

Property	Description	Default Value	Notes
topic.name	Name of the topic to create partitions for	None	Required field
partition.count	Number of partitions to create	1	Must be a positive integer
replication.factor	Replication factor for each partition	1	Must be less than or equal to number of brokers

Example: Creating Partitions via Admin Client

const { AdminClient } = require('kafkajs');

async function createPartitions() {
  const admin = AdminClient.create({
    brokers: ['localhost:9092'],
    clientId: 'partition-creator'
  });

  await admin.createPartitions({
    topicName: 'my-topic',
    count: 3
  });

  console.log('Partitions created successfully');
}

createPartitions().catch(console.error);

Note: The number of partitions can only be increased, not decreased.

7. Altering Configs

Config Type	Resource	Properties	Notes
Broker Config	`broker.id`	`1001`	Unique identifier for the broker
Broker Config	`log.dirs`	`/var/lib/kafka/data`	Directories for log storage
Topic Config	`retention.ms`	`604800000`	Retention period in milliseconds (7 days)
Topic Config	`replication.factor`	`3`	Number of replicas for each partition
Consumer Config	`session.timeout.ms`	`45000`	Timeout for consumer session

Example: Altering Topic Configs

const { AdminClient } = require("kafkajs");

const admin = AdminClient.create({
  brokers: ["localhost:9092"],
});

await admin.setConfigs({
  resources: [
    {
      type: 2, // Topic
      name: "my-topic",
      configs: [
        { name: "retention.ms", value: "1209600000" },
        { name: "segment.bytes", value: "1073741824" },
      ],
    },
  ],
});

Note: Altering configs requires appropriate permissions and may require a restart of the affected brokers or topics.

8. Describing Configs

Config Type	Resource	Description	Notes
broker	`broker`	Broker-level configuration	Applies to all brokers in the cluster
topic	`topic`	Topic-level configuration	Applies to a specific topic
user	`user`	User-level configuration	Applies to a specific user
client	`client`	Client-level configuration	Applies to a specific client

Example: Describe Broker Configs

kafka-configs.sh --bootstrap-server localhost:9092 --entity-type broker --entity-name 0 --describe

Example: Describe Topic Configs

kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topic --entity-name my-topic --describe

Note: The kafka-configs.sh script is used to describe configurations for various Kafka entities.

9. Describing Consumer Groups

Property	Description	Values
group.id	Unique identifier for the consumer group	`my-consumer-group`
session.timeout.ms	Timeout for consumer session	`45000`
heartbeat.interval.ms	Interval between heartbeats	`3000`

Example: Describe Consumer Group

const { AdminClient } = require('kafkajs');

async function describeConsumerGroups() {
  const admin = AdminClient.create({
    brokers: ['localhost:9092'],
    clientId: 'admin-client'
  });

  const result = await admin.describeConsumerGroups([
    { groupId: 'my-consumer-group' }
  ]);

  console.log(result);
}

describeConsumerGroups().catch(console.error);

Note: The Admin API provides methods to describe consumer groups, including their members, states, and assigned partitions.

10. Listing Consumer Groups

Property	Description	Default Value	Notes
group.id	Unique identifier for the consumer group	None	Required field when creating a consumer
session.timeout.ms	Timeout for consumer session	45000	Consumer must send heartbeats within this time
heartbeat.interval.ms	Interval between heartbeats	3000	Must be less than session.timeout.ms

Example: Listing Consumer Groups via Admin Client

const { AdminClient } = require('kafkajs');

async function listConsumerGroups() {
  const admin = AdminClient.create({
    brokers: ['localhost:9092'],
    clientId: 'admin-client'
  });

  try {
    const groups = await admin.listConsumerGroups();
    console.log('Consumer Groups:', groups);
  } catch (error) {
    console.error('Error listing consumer groups:', error);
  }
}

listConsumerGroups();

Note: The AdminClient must be connected to a running Kafka cluster to list consumer groups.

11. Deleting Consumer Groups

Property	Description	Notes
group.id	Identifier of the consumer group to delete	Required parameter for deletion

Example: Delete Consumer Group

kafka-consumer-groups.sh --bootstrap-server localhost:9092 --delete --group my-group

Note: Consumer groups must be inactive (no active consumers) to be deleted.

Warning: Deleting a consumer group will remove all its offsets from the __consumer_offsets topic.

12. Deleting Records

Deleting records in Kafka is achieved through the deleteRecords operation, which allows for the removal of messages from a topic up to a specified offset. This operation is part of the Admin API and requires appropriate permissions.

Parameter	Description	Required
topic	Name of the topic from which records are to be deleted	Yes
offsets	Map of partition numbers to the offset up to which records should be deleted	Yes

Example: Deleting Records from a Topic

const { Kafka } = require('kafkajs');

const kafka = new Kafka({
  clientId: 'my-app',
  brokers: ['localhost:9092']
});

const admin = kafka.admin();

await admin.connect();

const result = await admin.deleteRecords({
  topic: 'my-topic',
  offsets: {
    0: 100,
    1: 200
  }
});

console.log('Records deleted:', result);

Note: The deleteRecords operation does not immediately remove the data from disk. It marks the records as deleted and they will be removed during log compaction or retention policy enforcement.

Warning: Deleting records can lead to data inconsistency if consumers rely on the order or presence of messages. Ensure that all consumers are aware of and can handle such deletions.