Skip to content

Commit d08cec4

Browse files
awlaytonawlayton
committed
feat(kafka): attempt to create topics on startup
1 parent 62c180e commit d08cec4

File tree

16 files changed

+173
-77
lines changed

16 files changed

+173
-77
lines changed

‎oada/.eslintrc.yaml

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@ extends:
1919
plugins:
2020
- '@typescript-eslint'
2121
- node
22+
- escompat
2223
- github
2324
- promise
2425
- regexp

‎oada/libs/lib-kafka/src/base.ts renamed to ‎oada/libs/lib-kafka/src/Base.ts

Lines changed: 10 additions & 50 deletions
Original file line numberDiff line numberDiff line change
@@ -20,11 +20,12 @@ import { config } from './config.js';
2020
import { once } from 'node:events';
2121
import process from 'node:process';
2222

23-
import type { Consumer, EachMessagePayload, Producer, logLevel } from 'kafkajs';
23+
import type { Consumer, EachMessagePayload, Producer } from 'kafkajs';
2424
import EventEmitter from 'eventemitter3';
25-
import { Kafka } from 'kafkajs';
2625
import debug from 'debug';
2726

27+
import Kafka from './Kafka.js';
28+
2829
// Const info = debug('@oada/lib-kafka:info');
2930
const error = debug('@oada/lib-kafka:error');
3031

@@ -55,7 +56,7 @@ function die(reason: Error) {
5556
}
5657

5758
export interface ConstructorOptions {
58-
consumeTopic: string | string[];
59+
consumeTopic: string | readonly string[];
5960
// eslint-disable-next-line @typescript-eslint/ban-types
6061
produceTopic?: string | null;
6162
group: string;
@@ -86,34 +87,14 @@ export interface KafkaBase {
8687
domain?: string;
8788
}
8889

89-
/**
90-
* Make kafkajs logging nicer?
91-
*/
92-
type KafkajsDebug = Record<
93-
keyof Omit<typeof logLevel, 'NOTHING'>,
94-
debug.Debugger
95-
>;
96-
const kafkajsDebugs = new Map<string, KafkajsDebug>();
97-
function getKafkajsDebug(namespace: string): KafkajsDebug {
98-
const d = kafkajsDebugs.get(namespace);
99-
if (d) {
100-
return d;
101-
}
102-
103-
const newDebug = {
104-
ERROR: debug(`kafkajs:${namespace}:error`),
105-
WARN: debug(`kafkajs:${namespace}:warn`),
106-
INFO: debug(`kafkajs:${namespace}:info`),
107-
DEBUG: debug(`kafkajs:${namespace}:debug`),
108-
};
109-
kafkajsDebugs.set(namespace, newDebug);
110-
return newDebug;
90+
function isArray(value: unknown): value is unknown[] | readonly unknown[] {
91+
return Array.isArray(value);
11192
}
11293

11394
export class Base extends EventEmitter {
11495
protected static done = Symbol('kafka-base-done');
11596

116-
readonly consumeTopic;
97+
readonly consumeTopics;
11798
readonly produceTopic;
11899
readonly group;
119100
readonly #kafka: Kafka;
@@ -130,29 +111,11 @@ export class Base extends EventEmitter {
130111
}: ConstructorOptions) {
131112
super();
132113

133-
this.consumeTopic = consumeTopic;
114+
this.consumeTopics = isArray(consumeTopic) ? consumeTopic : [consumeTopic];
134115
this.produceTopic = produceTopic;
135116
this.group = group;
136117

137-
this.#kafka = new Kafka({
138-
/**
139-
* Make kafkajs logging nicer?
140-
*/
141-
logCreator() {
142-
return ({ namespace, label, log }) => {
143-
const l = label as keyof KafkajsDebug;
144-
// eslint-disable-next-line security/detect-object-injection
145-
const logger = getKafkajsDebug(namespace)[l];
146-
if (log instanceof Error) {
147-
logger({ err: log }, log.message);
148-
} else {
149-
const { message, ...extra } = log;
150-
logger(extra, message);
151-
}
152-
};
153-
},
154-
brokers: config.get('kafka.broker'),
155-
});
118+
this.#kafka = new Kafka();
156119

157120
this.consumer =
158121
consumer ??
@@ -245,10 +208,7 @@ export class Base extends EventEmitter {
245208
await this.consumer.connect();
246209
await this.producer.connect();
247210

248-
for (const topic of Array.isArray(this.consumeTopic)
249-
? this.consumeTopic
250-
: [this.consumeTopic]) {
251-
// eslint-disable-next-line no-await-in-loop
211+
for await (const topic of this.consumeTopics) {
252212
await this.consumer.subscribe({ topic });
253213
}
254214

‎oada/libs/lib-kafka/src/Kafka.ts

Lines changed: 77 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,77 @@
1+
/**
2+
* @license
3+
* Copyright 2022 Open Ag Data Alliance
4+
*
5+
* Licensed under the Apache License, Version 2.0 (the "License");
6+
* you may not use this file except in compliance with the License.
7+
* You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
import { config } from './config.js';
19+
20+
import { Kafka, type KafkaConfig, type logLevel } from 'kafkajs';
21+
import debug from 'debug';
22+
23+
/**
24+
* Make kafkajs logging nicer?
25+
*/
26+
type KafkajsDebug = Record<
27+
keyof Omit<typeof logLevel, 'NOTHING'>,
28+
debug.Debugger
29+
>;
30+
const kafkajsDebugs = new Map<string, KafkajsDebug>();
31+
function getKafkajsDebug(namespace: string): KafkajsDebug {
32+
const d = kafkajsDebugs.get(namespace);
33+
if (d) {
34+
return d;
35+
}
36+
37+
const newDebug = {
38+
ERROR: debug(`kafkajs:${namespace}:error`),
39+
WARN: debug(`kafkajs:${namespace}:warn`),
40+
INFO: debug(`kafkajs:${namespace}:info`),
41+
DEBUG: debug(`kafkajs:${namespace}:debug`),
42+
};
43+
kafkajsDebugs.set(namespace, newDebug);
44+
return newDebug;
45+
}
46+
47+
/**
48+
* Wraps the `Kafka` client class to add our own defaults etc.
49+
* @see {@link Kafka}
50+
*/
51+
export default class IKafka extends Kafka {
52+
constructor({
53+
brokers = config.get('kafka.broker'),
54+
...rest
55+
}: Partial<KafkaConfig> = {}) {
56+
super({
57+
...rest,
58+
/**
59+
* Make kafkajs logging nicer?
60+
*/
61+
logCreator() {
62+
return ({ namespace, label, log }) => {
63+
const l = label as keyof KafkajsDebug;
64+
// eslint-disable-next-line security/detect-object-injection
65+
const logger = getKafkajsDebug(namespace)[l];
66+
if (log instanceof Error) {
67+
logger({ err: log }, log.message);
68+
} else {
69+
const { message, ...extra } = log;
70+
logger(extra, message);
71+
}
72+
};
73+
},
74+
brokers,
75+
});
76+
}
77+
}

‎oada/libs/lib-kafka/src/ReResponder.ts

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,7 @@
1515
* limitations under the License.
1616
*/
1717

18-
import { DATA, REQ_ID_KEY } from './base.js';
18+
import { DATA, REQ_ID_KEY } from './Base.js';
1919
import { Responder } from './Responder.js';
2020

2121
import ksuid from 'ksuid';

‎oada/libs/lib-kafka/src/Requester.ts

Lines changed: 10 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -21,8 +21,15 @@ import { setTimeout } from 'node:timers/promises';
2121
import EventEmitter from 'eventemitter3';
2222
import ksuid from 'ksuid';
2323

24-
import { Base, CANCEL_KEY, DATA, REQ_ID_KEY, topicTimeout } from './base.js';
25-
import type { ConstructorOptions, KafkaBase } from './base.js';
24+
import {
25+
Base,
26+
CANCEL_KEY,
27+
type ConstructorOptions,
28+
DATA,
29+
type KafkaBase,
30+
REQ_ID_KEY,
31+
topicTimeout,
32+
} from './Base.js';
2633

2734
export class KafkaRequestTimeoutError extends Error {}
2835

@@ -132,4 +139,4 @@ export class Requester extends Base {
132139
}
133140
}
134141

135-
export type { ConstructorOptions } from './base.js';
142+
export type { ConstructorOptions } from './Base.js';

‎oada/libs/lib-kafka/src/Responder.ts

Lines changed: 9 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -17,11 +17,15 @@
1717

1818
import util from 'node:util';
1919

20-
import { Base, CANCEL_KEY, DATA, REQ_ID_KEY, topicTimeout } from './base.js';
21-
import type {
22-
ConstructorOptions as BaseConstructorOptions,
23-
KafkaBase,
24-
} from './base.js';
20+
import {
21+
Base,
22+
type ConstructorOptions as BaseConstructorOptions,
23+
CANCEL_KEY,
24+
DATA,
25+
type KafkaBase,
26+
REQ_ID_KEY,
27+
topicTimeout,
28+
} from './Base.js';
2529

2630
import type { EachMessagePayload } from 'kafkajs';
2731
import debug from 'debug';

‎oada/libs/lib-kafka/src/ResponderRequester.ts

Lines changed: 11 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -18,15 +18,16 @@
1818
import type { EachMessagePayload } from 'kafkajs';
1919
import type EventEmitter from 'eventemitter3';
2020

21-
import { Base, DATA } from './base.js';
22-
import type {
23-
ConstructorOptions as ResponderOptions,
24-
Response,
21+
import { Base, DATA, type KafkaBase } from './Base.js';
22+
import {
23+
Requester,
24+
type ConstructorOptions as RequesterOptions,
25+
} from './Requester.js';
26+
import {
27+
Responder,
28+
type ConstructorOptions as ResponderOptions,
29+
type Response,
2530
} from './Responder.js';
26-
import type { KafkaBase } from './base.js';
27-
import { Requester } from './Requester.js';
28-
import type { ConstructorOptions as RequesterOptions } from './Requester.js';
29-
import { Responder } from './Responder.js';
3031

3132
import debug from 'debug';
3233

@@ -109,12 +110,12 @@ export class ResponderRequester extends Base {
109110
// Mux the consumer between requester and responder
110111
this.on(DATA, (value: KafkaBase, data, ...rest) => {
111112
trace(data, 'Received data: %o', value);
112-
if (data.topic === this.#requester.consumeTopic) {
113+
if (this.#requester.consumeTopics.includes(data.topic)) {
113114
trace('Muxing data to requester');
114115
this.#requester.emit(DATA, value, data, ...rest);
115116
}
116117

117-
if (data.topic === this.#responder.consumeTopic) {
118+
if (this.#responder.consumeTopics.includes(data.topic)) {
118119
if (!this.#respondOwn && value.group === this.group) {
119120
// Don't respond to own requests
120121
return;

‎oada/libs/lib-kafka/src/index.ts

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,8 @@
1818
// @ts-expect-error the types are not correct
1919
export { KafkaJSError as KafkaError } from 'kafkajs/src/errors.js';
2020

21-
export type { KafkaBase } from './base.js';
21+
export * as init from './init.js';
22+
export type { KafkaBase } from './Base.js';
2223
export { Responder } from './Responder.js';
2324
export { ReResponder } from './ReResponder.js';
2425
export { Requester } from './Requester.js';

‎oada/libs/lib-kafka/src/init.ts

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,39 @@
1+
/**
2+
* @license
3+
* Copyright 2022 Open Ag Data Alliance
4+
*
5+
* Licensed under the Apache License, Version 2.0 (the "License");
6+
* you may not use this file except in compliance with the License.
7+
* You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
import { config } from './config.js';
19+
20+
import Kafka from './Kafka.js';
21+
22+
/**
23+
* Ensure our Kafka topics exist
24+
*/
25+
export async function run(): Promise<void> {
26+
const kafka = new Kafka();
27+
const topics = config.get('kafka.topics');
28+
29+
const admin = kafka.admin();
30+
await admin.connect();
31+
try {
32+
await admin.createTopics({
33+
waitForLeaders: false,
34+
topics: Object.values(topics).map((topic) => ({ topic })),
35+
});
36+
} finally {
37+
await admin.disconnect();
38+
}
39+
}

‎oada/libs/lib-kafka/test/Requester.test.ts

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,7 @@ import { Kafka } from 'kafkajs';
2626
import { v4 as uuid } from 'uuid';
2727

2828
import { KafkaRequestTimeoutError, Requester } from '../dist/Requester.js';
29-
import type { KafkaBase } from '../src/base.js';
29+
import type { KafkaBase } from '../src/Base.js';
3030

3131
const REQ_TOPIC = 'test_requests';
3232
const RES_TOPIC = 'test_responses';

‎oada/libs/lib-kafka/test/Responder.test.ts

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,7 @@ import type { Consumer, Producer } from 'kafkajs';
2525
import { Kafka } from 'kafkajs';
2626
import { v4 as uuid } from 'uuid';
2727

28-
import type { KafkaBase } from '../src/base.js';
28+
import type { KafkaBase } from '../src/Base.js';
2929
import { Responder } from '../dist/Responder.js';
3030

3131
const REQ_TOPIC = 'test_requests';

‎oada/package.json

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@
3030
"eslint-import-resolver-node": "^0.3.6",
3131
"eslint-plugin-array-func": "^3.1.7",
3232
"eslint-plugin-ava": "^13.2.0",
33+
"eslint-plugin-escompat": "^3.3.3",
3334
"eslint-plugin-eslint-comments": "^3.2.0",
3435
"eslint-plugin-filenames": "^1.3.2",
3536
"eslint-plugin-github": "^4.4.0",

‎oada/services/startup/package.json

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@
3030
"homepage": "https://github.com/oada/oada-srvc-startup#readme",
3131
"dependencies": {
3232
"@oada/lib-arangodb": "^3.5.1",
33+
"@oada/lib-kafka": "^3.5.1",
3334
"@oada/pino-debug": "^3.5.1",
3435
"debug": "^4.3.4",
3536
"tslib": "^2.4.0"

‎oada/services/startup/src/index.ts

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,8 @@
1717

1818
import http from 'node:http';
1919

20-
import { init } from '@oada/lib-arangodb';
20+
import { init as initArangoDB } from '@oada/lib-arangodb';
21+
import { init as initKafka } from '@oada/lib-kafka';
2122

2223
import debug from 'debug';
2324

@@ -27,9 +28,9 @@ const info = debug('startup:info');
2728
const port = process.env.PORT ?? 8080;
2829
const exit = process.env.EXIT ?? false;
2930

30-
info('Startup is creating the database');
31-
await init.run();
32-
info('Database created/ensured.');
31+
info('Startup is initializing ArangoDB and Kafka');
32+
await Promise.all([initArangoDB.run(), initKafka.run()]);
33+
info('Initialization complete');
3334

3435
if (exit) {
3536
// eslint-disable-next-line no-process-exit, unicorn/no-process-exit

0 commit comments

Comments
 (0)