Skip to content

Commit d014984

Browse files
emasabemasab
authored
Admin examples for available APIs (#84)
1 parent 8ce7cfb commit d014984

File tree

7 files changed

+412
-40
lines changed

7 files changed

+412
-40
lines changed

‎examples/kafkajs/admin.js

Lines changed: 0 additions & 40 deletions
This file was deleted.

‎examples/kafkajs/admin/create-topics.js

Lines changed: 79 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,79 @@
1+
// require('kafkajs') is replaced with require('@confluentinc/kafka-javascript').KafkaJS.
2+
const { Kafka } = require('@confluentinc/kafka-javascript').KafkaJS;
3+
const { parseArgs } = require('node:util');
4+
5+
async function adminStart() {
6+
const args = parseArgs({
7+
options: {
8+
'bootstrap-servers': {
9+
type: 'string',
10+
short: 'b',
11+
default: 'localhost:9092',
12+
},
13+
'topic': {
14+
type: 'string',
15+
short: 't',
16+
default: 'test-topic',
17+
},
18+
'timeout': {
19+
type: 'string',
20+
short: 'm',
21+
default: undefined,
22+
},
23+
'num-partitions': {
24+
type: 'string',
25+
short: 'p',
26+
default: '3',
27+
},
28+
'replication-factor': {
29+
type: 'string',
30+
short: 'r',
31+
default: '1',
32+
}
33+
},
34+
});
35+
36+
let {
37+
'bootstrap-servers': bootstrapServers,
38+
timeout,
39+
'num-partitions': numPartitions,
40+
'replication-factor': replicationFactor,
41+
topic,
42+
} = args.values;
43+
44+
if (timeout) {
45+
timeout = Number(timeout) || 0;
46+
}
47+
48+
numPartitions = Number(numPartitions) || 3;
49+
replicationFactor = Number(replicationFactor) || 1;
50+
51+
const kafka = new Kafka({
52+
kafkaJS: {
53+
brokers: [bootstrapServers],
54+
}
55+
});
56+
57+
const admin = kafka.admin();
58+
await admin.connect();
59+
60+
try {
61+
await admin.createTopics({
62+
topics: [
63+
{
64+
topic: topic,
65+
numPartitions: numPartitions,
66+
replicationFactor: replicationFactor,
67+
}
68+
],
69+
timeout,
70+
});
71+
console.log(`Topic "${topic}" created successfully`);
72+
} catch(err) {
73+
console.log(`Topic creation failed`, err);
74+
}
75+
76+
await admin.disconnect();
77+
}
78+
79+
adminStart();

‎examples/kafkajs/admin/delete-groups.js

Lines changed: 64 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,64 @@
1+
// require('kafkajs') is replaced with require('@confluentinc/kafka-javascript').KafkaJS.
2+
const { Kafka } = require('@confluentinc/kafka-javascript').KafkaJS;
3+
const { parseArgs } = require('node:util');
4+
5+
async function adminStart() {
6+
const args = parseArgs({
7+
options: {
8+
'bootstrap-servers': {
9+
type: 'string',
10+
short: 'b',
11+
default: 'localhost:9092',
12+
},
13+
'timeout': {
14+
type: 'string',
15+
short: 'm',
16+
default: undefined,
17+
},
18+
'group-ids': {
19+
type: 'string',
20+
short: 'g',
21+
multiple: true,
22+
default: [],
23+
},
24+
},
25+
});
26+
27+
let {
28+
'bootstrap-servers': bootstrapServers,
29+
timeout,
30+
'group-ids': groupIds,
31+
} = args.values;
32+
33+
if (!groupIds.length) {
34+
console.error('Group ids are required');
35+
process.exit(1);
36+
}
37+
38+
if (timeout) {
39+
timeout = Number(timeout) || 0;
40+
}
41+
42+
const kafka = new Kafka({
43+
kafkaJS: {
44+
brokers: [bootstrapServers],
45+
}
46+
});
47+
48+
const admin = kafka.admin();
49+
await admin.connect();
50+
51+
try {
52+
await admin.deleteGroups(
53+
groupIds,
54+
{ timeout },
55+
);
56+
console.log(`Groups "${groupIds.join(',')}" deleted successfully`);
57+
} catch(err) {
58+
console.log(`Group deletion failed`, err);
59+
}
60+
61+
await admin.disconnect();
62+
}
63+
64+
adminStart();

‎examples/kafkajs/admin/delete-topics.js

Lines changed: 64 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,64 @@
1+
// require('kafkajs') is replaced with require('@confluentinc/kafka-javascript').KafkaJS.
2+
const { Kafka } = require('@confluentinc/kafka-javascript').KafkaJS;
3+
const { parseArgs } = require('node:util');
4+
5+
async function adminStart() {
6+
const args = parseArgs({
7+
options: {
8+
'bootstrap-servers': {
9+
type: 'string',
10+
short: 'b',
11+
default: 'localhost:9092',
12+
},
13+
'timeout': {
14+
type: 'string',
15+
short: 'm',
16+
default: undefined,
17+
},
18+
'topics': {
19+
type: 'string',
20+
short: 't',
21+
multiple: true,
22+
default: [],
23+
},
24+
},
25+
});
26+
27+
let {
28+
'bootstrap-servers': bootstrapServers,
29+
timeout,
30+
topics,
31+
} = args.values;
32+
33+
if (!topics.length) {
34+
console.error('Topics names is required');
35+
process.exit(1);
36+
}
37+
38+
if (timeout) {
39+
timeout = Number(timeout) || 0;
40+
}
41+
42+
const kafka = new Kafka({
43+
kafkaJS: {
44+
brokers: [bootstrapServers],
45+
}
46+
});
47+
48+
const admin = kafka.admin();
49+
await admin.connect();
50+
51+
try {
52+
await admin.deleteTopics({
53+
topics,
54+
timeout,
55+
});
56+
console.log(`Topics "${topics.join(',')}" deleted successfully`);
57+
} catch(err) {
58+
console.log(`Topic deletion failed`, err);
59+
}
60+
61+
await admin.disconnect();
62+
}
63+
64+
adminStart();

‎examples/kafkajs/admin/describe-groups.js

Lines changed: 88 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,88 @@
1+
// require('kafkajs') is replaced with require('@confluentinc/kafka-javascript').KafkaJS.
2+
const { Kafka, ConsumerGroupStates } = require('@confluentinc/kafka-javascript').KafkaJS;
3+
const { parseArgs } = require('node:util');
4+
5+
function printNode(node, prefix = '') {
6+
if (!node)
7+
return;
8+
console.log(`${prefix}\tHost: ${node.host}`);
9+
console.log(`${prefix}\tPort: ${node.port}`);
10+
console.log(`${prefix}\tRack: ${node.rack}`);
11+
}
12+
13+
async function adminStart() {
14+
const args = parseArgs({
15+
options: {
16+
'bootstrap-servers': {
17+
type: 'string',
18+
short: 'b',
19+
default: 'localhost:9092',
20+
},
21+
'timeout': {
22+
type: 'string',
23+
short: 'm',
24+
default: undefined,
25+
},
26+
'groups': {
27+
type: 'string',
28+
short: 'g',
29+
multiple: true,
30+
default: [],
31+
},
32+
'include-authorized-operations': {
33+
type: 'boolean',
34+
short: 'i',
35+
default: false,
36+
}
37+
},
38+
});
39+
40+
let {
41+
'bootstrap-servers': bootstrapServers,
42+
timeout,
43+
groups,
44+
'include-authorized-operations': includeAuthorizedOperations,
45+
} = args.values;
46+
47+
if (timeout) {
48+
timeout = Number(timeout) || 0;
49+
}
50+
51+
const kafka = new Kafka({
52+
kafkaJS: {
53+
brokers: [bootstrapServers],
54+
}
55+
});
56+
57+
const admin = kafka.admin();
58+
await admin.connect();
59+
60+
try {
61+
const groupDescriptions = await admin.describeGroups(
62+
groups,
63+
{
64+
timeout,
65+
includeAuthorizedOperations,
66+
}
67+
);
68+
for (const group of groupDescriptions.groups) {
69+
console.log(`Group id: ${group.groupId}`);
70+
console.log(`\tError: ${group.error}`);
71+
console.log(`\tProtocol: ${group.protocol}`);
72+
console.log(`\tProtocol type: ${group.protocolType}`);
73+
console.log(`\tPartition assignor: ${group.partitionAssignor}`);
74+
console.log(`\tState: ${group.state}`);
75+
console.log(`\tCoordinator: ${group.coordinator ? group.coordinator.id : group.coordinator}`);
76+
printNode(group.coordinator, '\t');
77+
console.log(`\tAuthorized operations: ${group.authorizedOperations}`);
78+
console.log(`\tIs simple: ${group.isSimpleConsumerGroup}`);
79+
console.log(`\tState: ${group.state}`);
80+
}
81+
} catch(err) {
82+
console.log('Describe groups failed', err);
83+
}
84+
85+
await admin.disconnect();
86+
}
87+
88+
adminStart();

0 commit comments

Comments
 (0)