chriszs commented 4 years ago

Bug

Current Behavior

There's no connection or disconnect for the Kafka producer in AbstractIngestionEngine. This may contribute to application hangs.

Input

yarn babel-node src/components/ingestion/test.js

// src/components/ingestion/test.js
import 'module-alias/register'
import AbstractIngestionEngine from './AbstractIngestionEngine'

class FileIngestionEngine extends AbstractIngestionEngine {
    constructor(path) {
        super()

        this.path = path
    }

    path = null

    getInputStream = () => fs.createReadStream(this.path)
}

const engine = new FileIngestionEngine(__dirname + '/in.ts')

engine.start()

Expected Behavior

I expect it to connect and disconnect from the Kafka producer.

Possible Solutions

I added await this.producer.connect() in start, which I made async, and in stop:

if (this.producer !== null) {
    this.producer.disconnect()
}

Related Issues

57, #58, #59, #60

slifty commented 4 years ago

Oh ty -- also, full disclosure: there is a whole bunch we have punted on learning about kafkajs and the proper way to call / configure / etc with it. I suspect there's a bit of room for improvement to our /src/lib/kafka code worth exploring in its own issue sometime fairly soon.