Mirror of https://github.com/roostorg/coop
github.com/roostorg/coop
fix: Catch Kafka connect error to prevent unhandled promise rejection crashing the process (#96)
* fix: Catch Kafka connect error to prevent unhandled promise rejection crashing the process
* fix: lint issue
authored by
Caleb McQuaid
and committed by
GitHub
2ef16fda
f19196ef
+9
-1
+9
-1
server/kafka/itemQueueBulkWrite.ts
+9
-1
server/kafka/itemQueueBulkWrite.ts
···
21
21
topic: ITEM_SUBMISSION_SCHEMAS,
22
22
) {
23
23
const kafkaProducer = kafka.producer();
24
-
const initialConnectPromise = kafkaProducer.connect();
24
+
let connectError: Error | undefined;
25
+
const initialConnectPromise = kafkaProducer.connect().catch((err: unknown) => {
26
+
// Store the error to prevent an unhandled promise rejection from crashing
27
+
// the process. We re-throw it when callers attempt to write to Kafka.
28
+
connectError = err instanceof Error ? err : new Error(String(err));
29
+
});
25
30
const batchTimeout = 500;
26
31
27
32
const loader: DataLoader<ItemSubmissionKafkaMessageValue, void> =
···
45
50
skipBatch: boolean = false,
46
51
) {
47
52
await initialConnectPromise;
53
+
if (connectError) {
54
+
throw connectError;
55
+
}
48
56
// bulkWrite and loader.loadMany have different return types, so we have to
49
57
// handle their returns separately and construct a homogenous return type in
50
58
// each case, in addition to the logical difference of using batching or not