From 57b739433ccaa17f0bf184d77d4187d456f367e9 Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Wed, 3 Feb 2021 22:49:39 +0100 Subject: [PATCH 1/7] ingest to clickhouse only if PLUGIN_SERVER_INGESTION is enabled --- src/worker/queue.ts | 22 ++++++++++++---------- 1 file changed, 12 insertions(+), 10 deletions(-) diff --git a/src/worker/queue.ts b/src/worker/queue.ts index f55f1995..4cf943ab 100644 --- a/src/worker/queue.ts +++ b/src/worker/queue.ts @@ -92,16 +92,18 @@ async function startQueueKafka( status.error('❓', 'UUID missing in event received from Kafka!') return } - await server.eventsProcessor.processEvent( - distinct_id, - ip, - site_url, - event, - team_id, - DateTime.fromISO(now), - sent_at ? DateTime.fromISO(sent_at) : null, - uuid - ) + if (server.PLUGIN_SERVER_INGESTION) { + await server.eventsProcessor.processEvent( + distinct_id, + ip, + site_url, + event, + team_id, + DateTime.fromISO(now), + sent_at ? DateTime.fromISO(sent_at) : null, + uuid + ) + } }) await kafkaQueue.start() From 0bd1dbc546f839a4904e256c3066404fddddd0fd Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Wed, 3 Feb 2021 23:11:52 +0100 Subject: [PATCH 2/7] KAFKA_INCOMING_TOPIC --- src/config.ts | 2 ++ src/extensions/posthog.ts | 3 +-- src/ingestion/kafka-queue.ts | 9 ++++----- src/server.ts | 9 +++++++++ src/types.ts | 1 + tests/clickhouse/e2e.test.ts | 2 ++ tests/clickhouse/process-event.test.ts | 3 +++ 7 files changed, 22 insertions(+), 7 deletions(-) diff --git a/src/config.ts b/src/config.ts index d4846552..94c31962 100644 --- a/src/config.ts +++ b/src/config.ts @@ -20,6 +20,7 @@ export function getDefaultConfig(): PluginsServerConfig { KAFKA_CLIENT_CERT_B64: null, KAFKA_CLIENT_CERT_KEY_B64: null, KAFKA_TRUSTED_CERT_B64: null, + KAFKA_INCOMING_TOPIC: null, PLUGIN_SERVER_INGESTION: false, PLUGINS_CELERY_QUEUE: 'posthog-plugins', REDIS_URL: 'redis://127.0.0.1', @@ -62,6 +63,7 @@ export function getConfigHelp(): Record { LOG_LEVEL: 'minimum log level', KAFKA_ENABLED: 'use Kafka instead of Celery to ingest events', KAFKA_HOSTS: 'comma-delimited Kafka hosts', + KAFKA_INCOMING_TOPIC: 'set this override the kafka topic for incoming events', KAFKA_CLIENT_CERT_B64: 'Kafka certificate in Base64', KAFKA_CLIENT_CERT_KEY_B64: 'Kafka certificate key in Base64', KAFKA_TRUSTED_CERT_B64: 'Kafka trusted CA in Base64', diff --git a/src/extensions/posthog.ts b/src/extensions/posthog.ts index 94af08ca..2ee1a094 100644 --- a/src/extensions/posthog.ts +++ b/src/extensions/posthog.ts @@ -1,4 +1,3 @@ -import { KAFKA_EVENTS_INGESTION_HANDOFF } from '../ingestion/topics' import { Properties } from '@posthog/plugin-scaffold' import { DateTime } from 'luxon' import { PluginsServer, PluginConfig, RawEventMessage } from 'types' @@ -31,7 +30,7 @@ export function createPosthog(server: PluginsServer, pluginConfig: PluginConfig) throw new Error('kafkaProducer not configured!') } server.kafkaProducer.send({ - topic: KAFKA_EVENTS_INGESTION_HANDOFF, + topic: server.KAFKA_INCOMING_TOPIC!, messages: [ { key: data.uuid, diff --git a/src/ingestion/kafka-queue.ts b/src/ingestion/kafka-queue.ts index 26d1cc69..ec63c1fa 100644 --- a/src/ingestion/kafka-queue.ts +++ b/src/ingestion/kafka-queue.ts @@ -1,7 +1,6 @@ import * as Sentry from '@sentry/node' import { Kafka, Consumer, Message, EachBatchPayload } from 'kafkajs' import { PluginsServer, Queue, RawEventMessage } from 'types' -import { KAFKA_EVENTS_INGESTION_HANDOFF } from './topics' import { PluginEvent } from '@posthog/plugin-scaffold' import { status } from '../status' import { killGracefully } from '../utils' @@ -85,7 +84,7 @@ export class KafkaQueue implements Queue { this.consumer.on(this.consumer.events.CRASH, ({ payload: { error } }) => reject(error)) status.info('⏬', `Connecting Kafka consumer to ${this.pluginsServer.KAFKA_HOSTS}...`) this.wasConsumerRan = true - await this.consumer.subscribe({ topic: KAFKA_EVENTS_INGESTION_HANDOFF }) + await this.consumer.subscribe({ topic: this.pluginsServer.KAFKA_INCOMING_TOPIC! }) // KafkaJS batching: https://kafka.js.org/docs/consuming#a-name-each-batch-a-eachbatch await this.consumer.run({ // TODO: eachBatchAutoResolve: false, // don't autoresolve whole batch in case we exit it early @@ -104,7 +103,7 @@ export class KafkaQueue implements Queue { return } status.info('⏳', 'Pausing Kafka consumer...') - await this.consumer.pause([{ topic: KAFKA_EVENTS_INGESTION_HANDOFF }]) + await this.consumer.pause([{ topic: this.pluginsServer.KAFKA_INCOMING_TOPIC! }]) status.info('⏸', 'Kafka consumer paused!') } @@ -113,12 +112,12 @@ export class KafkaQueue implements Queue { return } status.info('⏳', 'Resuming Kafka consumer...') - await this.consumer.resume([{ topic: KAFKA_EVENTS_INGESTION_HANDOFF }]) + await this.consumer.resume([{ topic: this.pluginsServer.KAFKA_INCOMING_TOPIC! }]) status.info('▶️', 'Kafka consumer resumed!') } isPaused(): boolean { - return this.consumer.paused().some(({ topic }) => topic === KAFKA_EVENTS_INGESTION_HANDOFF) + return this.consumer.paused().some(({ topic }) => topic === this.pluginsServer.KAFKA_INCOMING_TOPIC) } async stop(): Promise { diff --git a/src/server.ts b/src/server.ts index 2ff92100..c9b54108 100644 --- a/src/server.ts +++ b/src/server.ts @@ -20,6 +20,7 @@ import { startSchedule } from './services/schedule' import { ConnectionOptions } from 'tls' import { DB } from './db' import { DateTime } from 'luxon' +import { KAFKA_EVENTS_INGESTION_HANDOFF, KAFKA_EVENTS_WAL } from './ingestion/topics' export async function createServer( config: Partial = {}, @@ -80,6 +81,14 @@ export async function createServer( database: serverConfig.CLICKHOUSE_DATABASE, }, }) + + if (!serverConfig.KAFKA_INCOMING_TOPIC) { + // When ingesting events, listen to the "INGESTION_HANDOFF" topic, otherwise listen to the "WAL" and discard + serverConfig.KAFKA_INCOMING_TOPIC = serverConfig.PLUGIN_SERVER_INGESTION + ? KAFKA_EVENTS_INGESTION_HANDOFF + : KAFKA_EVENTS_WAL + } + kafka = new Kafka({ clientId: `plugin-server-v${version}-${new UUIDT()}`, brokers: serverConfig.KAFKA_HOSTS.split(','), diff --git a/src/types.ts b/src/types.ts index 97bfef79..fa37223b 100644 --- a/src/types.ts +++ b/src/types.ts @@ -34,6 +34,7 @@ export interface PluginsServerConfig extends Record { KAFKA_CLIENT_CERT_B64: string | null KAFKA_CLIENT_CERT_KEY_B64: string | null KAFKA_TRUSTED_CERT_B64: string | null + KAFKA_INCOMING_TOPIC: string | null PLUGINS_CELERY_QUEUE: string REDIS_URL: string BASE_DIR: string diff --git a/tests/clickhouse/e2e.test.ts b/tests/clickhouse/e2e.test.ts index 1bd7e7fc..6ecfc1ed 100644 --- a/tests/clickhouse/e2e.test.ts +++ b/tests/clickhouse/e2e.test.ts @@ -9,6 +9,7 @@ import { delay, UUIDT } from '../../src/utils' import { resetTestDatabaseClickhouse } from '../helpers/clickhouse' import { resetKafka } from '../helpers/kafka' import { delayUntilEventIngested } from '../shared/process-event' +import { KAFKA_EVENTS_INGESTION_HANDOFF } from '../../src/ingestion/topics' jest.setTimeout(60000) // 60 sec timeout @@ -17,6 +18,7 @@ const extraServerConfig: Partial = { KAFKA_HOSTS: process.env.KAFKA_HOSTS || 'kafka:9092', WORKER_CONCURRENCY: 2, PLUGIN_SERVER_INGESTION: true, + KAFKA_INCOMING_TOPIC: KAFKA_EVENTS_INGESTION_HANDOFF, LOG_LEVEL: LogLevel.Log, } diff --git a/tests/clickhouse/process-event.test.ts b/tests/clickhouse/process-event.test.ts index 7604e29d..ab23d6c4 100644 --- a/tests/clickhouse/process-event.test.ts +++ b/tests/clickhouse/process-event.test.ts @@ -2,12 +2,15 @@ import { PluginsServerConfig, Event } from '../../src/types' import { resetTestDatabaseClickhouse } from '../helpers/clickhouse' import { resetKafka } from '../helpers/kafka' import { createProcessEventTests } from '../shared/process-event' +import { KAFKA_EVENTS_INGESTION_HANDOFF } from '../../src/ingestion/topics' jest.setTimeout(180_000) // 3 minute timeout const extraServerConfig: Partial = { KAFKA_ENABLED: true, KAFKA_HOSTS: process.env.KAFKA_HOSTS || 'kafka:9092', + PLUGIN_SERVER_INGESTION: true, + KAFKA_INCOMING_TOPIC: KAFKA_EVENTS_INGESTION_HANDOFF, } describe('process event (clickhouse)', () => { From b761c8efec52e20b980e3d5cc02439050f8696b6 Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Wed, 3 Feb 2021 23:12:19 +0100 Subject: [PATCH 3/7] test clickhouse connection --- src/server.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/src/server.ts b/src/server.ts index c9b54108..000270b1 100644 --- a/src/server.ts +++ b/src/server.ts @@ -81,6 +81,7 @@ export async function createServer( database: serverConfig.CLICKHOUSE_DATABASE, }, }) + await clickhouse.query('SELECT 1') if (!serverConfig.KAFKA_INCOMING_TOPIC) { // When ingesting events, listen to the "INGESTION_HANDOFF" topic, otherwise listen to the "WAL" and discard From ef9da0967378deda7f2782e2f3246e3abf839d92 Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Wed, 3 Feb 2021 23:12:44 +0100 Subject: [PATCH 4/7] remove some unused code --- src/ingestion/utils.ts | 16 +----- tests/helpers/kafka.ts | 114 ++--------------------------------------- 2 files changed, 4 insertions(+), 126 deletions(-) diff --git a/src/ingestion/utils.ts b/src/ingestion/utils.ts index 5b4cbb2a..ba46ab3b 100644 --- a/src/ingestion/utils.ts +++ b/src/ingestion/utils.ts @@ -1,20 +1,6 @@ -import { DateTime } from 'luxon' -import { Element, BaseEventMessage, RawEventMessage, EventMessage, BasePerson, RawPerson, Person } from '../types' +import { Element, BasePerson, RawPerson, Person } from '../types' import crypto from 'crypto' -export function parseRawEventMessage(message: RawEventMessage): EventMessage { - return { - ...(message as BaseEventMessage), - data: JSON.parse(message.data), - now: DateTime.fromISO(message.now), - sent_at: DateTime.fromISO(message.sent_at), - } -} - -export function parseRawPerson(rawPerson: RawPerson): Person { - return { ...(rawPerson as BasePerson), created_at: DateTime.fromISO(rawPerson.created_at) } -} - export function unparsePersonPartial(person: Partial): Partial { return { ...(person as BasePerson), ...(person.created_at ? { created_at: person.created_at.toISO() } : {}) } } diff --git a/tests/helpers/kafka.ts b/tests/helpers/kafka.ts index 5195e85c..7d1b535a 100644 --- a/tests/helpers/kafka.ts +++ b/tests/helpers/kafka.ts @@ -1,116 +1,8 @@ -import { EventEmitter } from 'events' -import { Kafka, Consumer, logLevel, EachMessagePayload, Producer } from 'kafkajs' -import { - KAFKA_EVENTS, - KAFKA_EVENTS_INGESTION_HANDOFF, - KAFKA_SESSION_RECORDING_EVENTS, -} from '../../src/ingestion/topics' -import { parseRawEventMessage } from '../../src/ingestion/utils' -import { EventMessage, PluginsServerConfig } from '../../src/types' +import { Kafka, logLevel } from 'kafkajs' +import { PluginsServerConfig } from '../../src/types' import { delay, UUIDT } from '../../src/utils' import { defaultConfig, overrideWithEnv } from '../../src/config' - -export class KafkaObserver extends EventEmitter { - public kafka: Kafka - public producer: Producer - public consumer: Consumer - - private isStarted: boolean - - constructor(extraServerConfig: Partial) { - super() - const config = { ...overrideWithEnv(defaultConfig, process.env), ...extraServerConfig } - this.kafka = new Kafka({ - clientId: `plugin-server-test-${new UUIDT()}`, - brokers: (config.KAFKA_HOSTS || '').split(','), - logLevel: logLevel.WARN, - }) - this.producer = this.kafka.producer() - this.consumer = this.kafka.consumer({ - groupId: 'clickhouse-ingestion-test', - }) - this.isStarted = false - } - - public async start(): Promise { - console.info('observer started!') - if (this.isStarted) { - return - } - this.isStarted = true - return await new Promise(async (resolve, reject) => { - console.info('connecting producer') - await this.producer.connect() - console.info('subscribing consumer') - await this.consumer.subscribe({ topic: KAFKA_EVENTS }) - console.info('running consumer') - await this.consumer.run({ - eachMessage: async (payload) => { - console.info('message received!') - this.emit('message', payload) - }, - }) - console.info('setting group join and crash listeners') - const { CONNECT, GROUP_JOIN, CRASH } = this.consumer.events - this.consumer.on(CONNECT, () => { - console.log('consumer connected to kafka') - }) - this.consumer.on(GROUP_JOIN, () => { - console.log('joined group') - resolve() - }) - this.consumer.on(CRASH, ({ payload: { error } }) => reject(error)) - }) - } - - public async stop(): Promise { - this.removeAllListeners() - console.info('disconnecting producer') - await this.producer.disconnect() - console.info('stopping consumer') - await this.consumer.stop() - console.info('disconnecting consumer') - await this.consumer.disconnect() - } - - public async handOffMessage(message: EventMessage): Promise { - console.info('producing message') - await this.producer.send({ - topic: KAFKA_EVENTS_INGESTION_HANDOFF, - messages: [{ value: Buffer.from(JSON.stringify(message)) }], - }) - } -} - -export class KafkaCollector extends EventEmitter { - collection: EventMessage[] - kafkaObserver: KafkaObserver - - constructor(kafkaObserver: KafkaObserver) { - super() - this.collection = [] - this.kafkaObserver = kafkaObserver - kafkaObserver.addListener('message', (payload: EachMessagePayload) => { - console.info('message received') - this.collection.push(parseRawEventMessage(JSON.parse(payload.message.value!.toString()))) - this.emit('message') - }) - } - - async collect(numberOfMessages: number): Promise { - return await new Promise((resolve) => { - const resolveIfCollectedEnough = () => { - console.log('collection:', this.collection) - if (this.collection.length >= numberOfMessages) { - this.removeListener('message', resolveIfCollectedEnough) - resolve(this.collection) - } - } - this.addListener('message', resolveIfCollectedEnough) - resolveIfCollectedEnough() - }) - } -} +import { KAFKA_EVENTS_INGESTION_HANDOFF, KAFKA_SESSION_RECORDING_EVENTS } from '../../src/ingestion/topics' /** Clear the kafka queue */ export async function resetKafka(extraServerConfig: Partial, delayMs = 2000) { From b436bdedf93528ce046670ec2fe0509e648a7260 Mon Sep 17 00:00:00 2001 From: Michael Matloka Date: Thu, 4 Feb 2021 03:59:36 +0100 Subject: [PATCH 5/7] Replace _INCOMING_ with _CONSUMPTION_ --- src/config.ts | 4 ++-- src/extensions/posthog.ts | 2 +- src/ingestion/kafka-queue.ts | 8 ++++---- src/server.ts | 4 ++-- src/types.ts | 2 +- tests/clickhouse/e2e.test.ts | 2 +- tests/clickhouse/process-event.test.ts | 2 +- 7 files changed, 12 insertions(+), 12 deletions(-) diff --git a/src/config.ts b/src/config.ts index 94c31962..f0888db2 100644 --- a/src/config.ts +++ b/src/config.ts @@ -20,7 +20,7 @@ export function getDefaultConfig(): PluginsServerConfig { KAFKA_CLIENT_CERT_B64: null, KAFKA_CLIENT_CERT_KEY_B64: null, KAFKA_TRUSTED_CERT_B64: null, - KAFKA_INCOMING_TOPIC: null, + KAFKA_CONSUMPTION_TOPIC: null, PLUGIN_SERVER_INGESTION: false, PLUGINS_CELERY_QUEUE: 'posthog-plugins', REDIS_URL: 'redis://127.0.0.1', @@ -63,7 +63,7 @@ export function getConfigHelp(): Record { LOG_LEVEL: 'minimum log level', KAFKA_ENABLED: 'use Kafka instead of Celery to ingest events', KAFKA_HOSTS: 'comma-delimited Kafka hosts', - KAFKA_INCOMING_TOPIC: 'set this override the kafka topic for incoming events', + KAFKA_CONSUMPTION_TOPIC: 'Kafka consumption topic override', KAFKA_CLIENT_CERT_B64: 'Kafka certificate in Base64', KAFKA_CLIENT_CERT_KEY_B64: 'Kafka certificate key in Base64', KAFKA_TRUSTED_CERT_B64: 'Kafka trusted CA in Base64', diff --git a/src/extensions/posthog.ts b/src/extensions/posthog.ts index 2ee1a094..863d662d 100644 --- a/src/extensions/posthog.ts +++ b/src/extensions/posthog.ts @@ -30,7 +30,7 @@ export function createPosthog(server: PluginsServer, pluginConfig: PluginConfig) throw new Error('kafkaProducer not configured!') } server.kafkaProducer.send({ - topic: server.KAFKA_INCOMING_TOPIC!, + topic: server.KAFKA_CONSUMPTION_TOPIC!, messages: [ { key: data.uuid, diff --git a/src/ingestion/kafka-queue.ts b/src/ingestion/kafka-queue.ts index ec63c1fa..e8a95468 100644 --- a/src/ingestion/kafka-queue.ts +++ b/src/ingestion/kafka-queue.ts @@ -84,7 +84,7 @@ export class KafkaQueue implements Queue { this.consumer.on(this.consumer.events.CRASH, ({ payload: { error } }) => reject(error)) status.info('⏬', `Connecting Kafka consumer to ${this.pluginsServer.KAFKA_HOSTS}...`) this.wasConsumerRan = true - await this.consumer.subscribe({ topic: this.pluginsServer.KAFKA_INCOMING_TOPIC! }) + await this.consumer.subscribe({ topic: this.pluginsServer.KAFKA_CONSUMPTION_TOPIC! }) // KafkaJS batching: https://kafka.js.org/docs/consuming#a-name-each-batch-a-eachbatch await this.consumer.run({ // TODO: eachBatchAutoResolve: false, // don't autoresolve whole batch in case we exit it early @@ -103,7 +103,7 @@ export class KafkaQueue implements Queue { return } status.info('⏳', 'Pausing Kafka consumer...') - await this.consumer.pause([{ topic: this.pluginsServer.KAFKA_INCOMING_TOPIC! }]) + await this.consumer.pause([{ topic: this.pluginsServer.KAFKA_CONSUMPTION_TOPIC! }]) status.info('⏸', 'Kafka consumer paused!') } @@ -112,12 +112,12 @@ export class KafkaQueue implements Queue { return } status.info('⏳', 'Resuming Kafka consumer...') - await this.consumer.resume([{ topic: this.pluginsServer.KAFKA_INCOMING_TOPIC! }]) + await this.consumer.resume([{ topic: this.pluginsServer.KAFKA_CONSUMPTION_TOPIC! }]) status.info('▶️', 'Kafka consumer resumed!') } isPaused(): boolean { - return this.consumer.paused().some(({ topic }) => topic === this.pluginsServer.KAFKA_INCOMING_TOPIC) + return this.consumer.paused().some(({ topic }) => topic === this.pluginsServer.KAFKA_CONSUMPTION_TOPIC) } async stop(): Promise { diff --git a/src/server.ts b/src/server.ts index 000270b1..952dd747 100644 --- a/src/server.ts +++ b/src/server.ts @@ -83,9 +83,9 @@ export async function createServer( }) await clickhouse.query('SELECT 1') - if (!serverConfig.KAFKA_INCOMING_TOPIC) { + if (!serverConfig.KAFKA_CONSUMPTION_TOPIC) { // When ingesting events, listen to the "INGESTION_HANDOFF" topic, otherwise listen to the "WAL" and discard - serverConfig.KAFKA_INCOMING_TOPIC = serverConfig.PLUGIN_SERVER_INGESTION + serverConfig.KAFKA_CONSUMPTION_TOPIC = serverConfig.PLUGIN_SERVER_INGESTION ? KAFKA_EVENTS_INGESTION_HANDOFF : KAFKA_EVENTS_WAL } diff --git a/src/types.ts b/src/types.ts index fa37223b..5e644da9 100644 --- a/src/types.ts +++ b/src/types.ts @@ -34,7 +34,7 @@ export interface PluginsServerConfig extends Record { KAFKA_CLIENT_CERT_B64: string | null KAFKA_CLIENT_CERT_KEY_B64: string | null KAFKA_TRUSTED_CERT_B64: string | null - KAFKA_INCOMING_TOPIC: string | null + KAFKA_CONSUMPTION_TOPIC: string | null PLUGINS_CELERY_QUEUE: string REDIS_URL: string BASE_DIR: string diff --git a/tests/clickhouse/e2e.test.ts b/tests/clickhouse/e2e.test.ts index 6ecfc1ed..51ac85f2 100644 --- a/tests/clickhouse/e2e.test.ts +++ b/tests/clickhouse/e2e.test.ts @@ -18,7 +18,7 @@ const extraServerConfig: Partial = { KAFKA_HOSTS: process.env.KAFKA_HOSTS || 'kafka:9092', WORKER_CONCURRENCY: 2, PLUGIN_SERVER_INGESTION: true, - KAFKA_INCOMING_TOPIC: KAFKA_EVENTS_INGESTION_HANDOFF, + KAFKA_CONSUMPTION_TOPIC: KAFKA_EVENTS_INGESTION_HANDOFF, LOG_LEVEL: LogLevel.Log, } diff --git a/tests/clickhouse/process-event.test.ts b/tests/clickhouse/process-event.test.ts index ab23d6c4..41167ddf 100644 --- a/tests/clickhouse/process-event.test.ts +++ b/tests/clickhouse/process-event.test.ts @@ -10,7 +10,7 @@ const extraServerConfig: Partial = { KAFKA_ENABLED: true, KAFKA_HOSTS: process.env.KAFKA_HOSTS || 'kafka:9092', PLUGIN_SERVER_INGESTION: true, - KAFKA_INCOMING_TOPIC: KAFKA_EVENTS_INGESTION_HANDOFF, + KAFKA_CONSUMPTION_TOPIC: KAFKA_EVENTS_INGESTION_HANDOFF, } describe('process event (clickhouse)', () => { From ae5fd06e58a8f954acec25fe3365b457662b0d63 Mon Sep 17 00:00:00 2001 From: Michael Matloka Date: Thu, 4 Feb 2021 03:59:49 +0100 Subject: [PATCH 6/7] Add KAFKA_CONSUMPTION_TOPIC to README.md --- README.md | 1 + 1 file changed, 1 insertion(+) diff --git a/README.md b/README.md index a566067d..6d1e254b 100644 --- a/README.md +++ b/README.md @@ -47,6 +47,7 @@ There's a multitude of settings you can use to control the plugin server. Use th | CLICKHOUSE_SECURE | Secure ClickHouse connection | `false` | | KAFKA_ENABLED | use Kafka instead of Celery to ingest events | `false` | | KAFKA_HOSTS | comma-delimited Kafka hosts | `null` | +| KAFKA_CONSUMPTION_TOPIC | Kafka consumption topic override | `null` (automatic) | | KAFKA_CLIENT_CERT_B64 | Kafka certificate in Base64 | `null` | | KAFKA_CLIENT_CERT_KEY_B64 | Kafka certificate key in Base64 | `null` | | KAFKA_TRUSTED_CERT_B64 | Kafka trusted CA in Base64 | `null` | From 294a65729d91f4cbe20488a369b787933f33e96f Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Thu, 4 Feb 2021 08:50:32 +0100 Subject: [PATCH 7/7] add // --- src/server.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/server.ts b/src/server.ts index 952dd747..9442ce31 100644 --- a/src/server.ts +++ b/src/server.ts @@ -81,7 +81,7 @@ export async function createServer( database: serverConfig.CLICKHOUSE_DATABASE, }, }) - await clickhouse.query('SELECT 1') + await clickhouse.query('SELECT 1') // test that the connection works if (!serverConfig.KAFKA_CONSUMPTION_TOPIC) { // When ingesting events, listen to the "INGESTION_HANDOFF" topic, otherwise listen to the "WAL" and discard