Kill connected consumers of a cluster when it is deleted.

This commit is contained in:
Sage Vaillancourt 2022-08-29 07:33:25 -04:00
parent 3f98892ddd
commit e6dbc34e14
3 changed files with 29 additions and 14 deletions

26
app.js
View File

@ -95,7 +95,18 @@ router.post('/clusters', async (req, res, _next) => {
router.post('/clusters/delete', async (req, res, _next) => { router.post('/clusters/delete', async (req, res, _next) => {
const clusterName = req.body.clusterName const clusterName = req.body.clusterName
// TODO: Disconnect // Kill all consumers connected to this cluster and notify those clients via their websocket
consumers.forEach(([{ consumer, cluster }, socket]) => {
if (clusters[clusterName] !== cluster) {
// TODO This reference equality may not be sufficient?
return
}
killConsumer(consumer)
socket.send(JSON.stringify({
type: 'cluster_deleted',
message: clusterName,
}))
})
delete clusters[clusterName] delete clusters[clusterName]
delete config.clusters[clusterName] delete config.clusters[clusterName]
await storeConfig(config) await storeConfig(config)
@ -164,7 +175,7 @@ buildEscape(consumers, clusters)
wsServer.on('connection', socket => { wsServer.on('connection', socket => {
socket.send('CONNECTED') socket.send('CONNECTED')
socket.on('close', async () => { socket.on('close', async () => {
await killConsumer(consumers.get(socket)) await killConsumer(consumers.get(socket).consumer)
consumers.delete(socket) consumers.delete(socket)
}) })
@ -173,24 +184,27 @@ wsServer.on('connection', socket => {
if (message.mode === 'kill') { if (message.mode === 'kill') {
console.log('KILLING SOCKET') console.log('KILLING SOCKET')
await killConsumer(consumers.get(socket)) await killConsumer(consumers.get(socket).consumer)
consumers.delete(socket) consumers.delete(socket)
return return
} }
const currentMode = message.mode === 'realTime' ? realTimeSearch : search const currentMode = message.mode === 'realTime' ? realTimeSearch : search
const cluster = clusters[message.cluster] const cluster = clusters[message.cluster]
const run = async () => consumers.set(socket, await currentMode({ const run = async () => consumers.set(socket, {
consumer: await currentMode({
kafka: cluster.kafka, kafka: cluster.kafka,
searchCode: message.searchCode, searchCode: message.searchCode,
immutable: message.immutable, immutable: message.immutable,
topic: message.topic, topic: message.topic,
maxItems: message.maxItems, maxItems: message.maxItems,
socket socket
})) }),
cluster
})
run().catch(async e => { run().catch(async e => {
console.error('run() error occurred!', e.toString()) console.error('run() error occurred!', e.toString())
await killConsumer(consumers.get(socket)) await killConsumer(consumers.get(socket).consumer)
// Try again ONCE on failure // Try again ONCE on failure
run().catch(ee => socket.send('ERROR: ' + ee)) run().catch(ee => socket.send('ERROR: ' + ee))
}) })

View File

@ -3,7 +3,8 @@ const buildMatcher = code => new Function('message', 'value', code)
const deepCopy = obj => JSON.parse(JSON.stringify(obj)) const deepCopy = obj => JSON.parse(JSON.stringify(obj))
const realTimeMessageSearch = async ({ kafka, topic, onMessage, searchCode, immutable = true }) => { const realTimeMessageSearch = async ({ kafka, topic, onMessage, searchCode, immutable = true }) => {
const consumer = kafka.consumer({ groupId: 'sage-testing-group' }) // TODO Use a UUID/GUID instead
const consumer = kafka.consumer({ groupId: 'sage-testing-group' + Math.random() })
console.log({ topic, onMessage, searchCode, immutable }) console.log({ topic, onMessage, searchCode, immutable })
await consumer.connect() await consumer.connect()

View File

@ -4,7 +4,7 @@ const signalTraps = ['SIGTERM', 'SIGINT', 'SIGUSR2']
const disconnectAll = async (consumers, clusters) => { const disconnectAll = async (consumers, clusters) => {
await Promise.all([ await Promise.all([
...Object.values(clusters).map(async cluster => cluster.admin.disconnect()), ...Object.values(clusters).map(async cluster => cluster.admin.disconnect()),
...Array.from(consumers.values()).map(async consumer => consumer.disconnect()) ...Array.from(consumers.values()).map(async consumer => consumer.consumer.disconnect())
]) ])
} }