Commit c0187457 authored by Tobias Steiner's avatar Tobias Steiner

Remove partition

parent f5fc64d5
......@@ -58,7 +58,7 @@ kafka.init().then(async () => {
.pipe(new ReporterCounter({objectMode: true}, core.getReporter(10000, ), 'read'))
.pipe(new Extractor({objectMode: true}, core.getLogger()))
.pipe(new ReporterCounter({objectMode: true}, core.getReporter(), 'write'))
.pipe(new StreamProducerPreparer({objectMode: true, topic: core.getNconf().get('producer:geolinker:topics'), partition: (chunk) => chunk.provider, key: (d) => `${d.provider}${d.id}`}))
.pipe(new StreamProducerPreparer({objectMode: true, topic: core.getNconf().get('producer:geolinker:topics'), key: (d) => `${d.provider}${d.id}`}))
.pipe(new ProducerStream(geolinkerProducer, {objectMode: true, topic: core.getNconf().get('producer:geolinker:topics')}));
core.getLogger().info('Pipes are registered');
} catch (error) {
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment