From ee912bba5b7af7b251ac744d41e28641efaf7abd Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Wed, 3 Feb 2021 17:24:25 +0100 Subject: [PATCH 1/9] resolve all kafka offsets --- src/ingestion/kafka-queue.ts | 18 +++++++++++++++--- 1 file changed, 15 insertions(+), 3 deletions(-) diff --git a/src/ingestion/kafka-queue.ts b/src/ingestion/kafka-queue.ts index 26d1cc69..375c9c66 100644 --- a/src/ingestion/kafka-queue.ts +++ b/src/ingestion/kafka-queue.ts @@ -47,12 +47,13 @@ export class KafkaQueue implements Queue { ...rawEvent, data: JSON.parse(rawEvent.data), })) + const offsetMap = new Map() const pluginEvents: PluginEvent[] = rawEvents.map((rawEvent) => { - const { data: dataStr, ...restOfRawEvent } = rawEvent + const { data: dataStr, kafka_offset: kafkaOffset, ...restOfRawEvent } = rawEvent const event = { ...restOfRawEvent, ...JSON.parse(dataStr) } + offsetMap.set(event.uuid, kafkaOffset) return { ...event, - kafka_offset: restOfRawEvent.kafka_offset, site_url: event.site_url || null, ip: event.ip || null, } @@ -71,11 +72,22 @@ export class KafkaQueue implements Queue { } const singleIngestionTimer = new Date() await this.saveEvent(event) - resolveOffset(event.kafka_offset!) + if (event?.uuid) { + const offset = offsetMap.get(event.uuid) + if (offset) { + resolveOffset(offset) + offsetMap.delete(event.uuid) + } + } await heartbeat() await commitOffsetsIfNecessary() this.pluginsServer.statsd?.timing('kafka_queue.single_ingestion', singleIngestionTimer) } + for (const removedOffset of offsetMap.values()) { + resolveOffset(removedOffset) + await heartbeat() + await commitOffsetsIfNecessary() + } this.pluginsServer.statsd?.timing('kafka_queue.each_batch', batchProcessingTimer) } From 00c3478fc79231ca5199d06c25034b0b6c3f8181 Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Wed, 3 Feb 2021 17:26:23 +0100 Subject: [PATCH 2/9] remove a few lines --- src/ingestion/kafka-queue.ts | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) diff --git a/src/ingestion/kafka-queue.ts b/src/ingestion/kafka-queue.ts index 375c9c66..ffe54409 100644 --- a/src/ingestion/kafka-queue.ts +++ b/src/ingestion/kafka-queue.ts @@ -72,12 +72,10 @@ export class KafkaQueue implements Queue { } const singleIngestionTimer = new Date() await this.saveEvent(event) - if (event?.uuid) { - const offset = offsetMap.get(event.uuid) - if (offset) { - resolveOffset(offset) - offsetMap.delete(event.uuid) - } + const offset = event?.uuid ? offsetMap.get(event.uuid) : null + if (offset) { + resolveOffset(offset) + offsetMap.delete(event.uuid!) } await heartbeat() await commitOffsetsIfNecessary() From d160964911d1320499d78c7e9aac5526e3278764 Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Wed, 3 Feb 2021 21:42:11 +0100 Subject: [PATCH 3/9] clean up --- src/ingestion/kafka-queue.ts | 36 ++++++++---------------------------- src/plugins.ts | 2 +- 2 files changed, 9 insertions(+), 29 deletions(-) diff --git a/src/ingestion/kafka-queue.ts b/src/ingestion/kafka-queue.ts index ffe54409..99dd3e4f 100644 --- a/src/ingestion/kafka-queue.ts +++ b/src/ingestion/kafka-queue.ts @@ -6,14 +6,12 @@ import { PluginEvent } from '@posthog/plugin-scaffold' import { status } from '../status' import { killGracefully } from '../utils' -export type BatchCallback = (messages: Message[]) => Promise - export class KafkaQueue implements Queue { private pluginsServer: PluginsServer private kafka: Kafka private consumer: Consumer private wasConsumerRan: boolean - private processEventBatch: (batch: PluginEvent[]) => Promise + private processEventBatch: (batch: PluginEvent[]) => Promise private saveEvent: (event: PluginEvent) => Promise constructor( @@ -39,28 +37,18 @@ export class KafkaQueue implements Queue { isStale, }: EachBatchPayload): Promise { const batchProcessingTimer = new Date() - const rawEvents: RawEventMessage[] = batch.messages.map((message) => ({ - ...JSON.parse(message.value!.toString()), - kafka_offset: message.offset, - })) - const parsedEvents = rawEvents.map((rawEvent) => ({ - ...rawEvent, - data: JSON.parse(rawEvent.data), - })) - const offsetMap = new Map() - const pluginEvents: PluginEvent[] = rawEvents.map((rawEvent) => { - const { data: dataStr, kafka_offset: kafkaOffset, ...restOfRawEvent } = rawEvent - const event = { ...restOfRawEvent, ...JSON.parse(dataStr) } - offsetMap.set(event.uuid, kafkaOffset) + const pluginEvents = batch.messages.map((message) => { + const { data: dataStr, ...rawEvent } = JSON.parse(message.value!.toString()) + const event = { ...rawEvent, ...JSON.parse(dataStr) } return { ...event, site_url: event.site_url || null, ip: event.ip || null, } }) - const processedEvents: PluginEvent[] = ( - await this.processEventBatch(pluginEvents) - ).filter((event: PluginEvent[] | false | null | undefined) => Boolean(event)) + + const processedEvents = await this.processEventBatch(pluginEvents) + for (const event of processedEvents) { if (!isRunning()) { status.info('😮', 'Consumer not running anymore, canceling batch processing!') @@ -75,17 +63,11 @@ export class KafkaQueue implements Queue { const offset = event?.uuid ? offsetMap.get(event.uuid) : null if (offset) { resolveOffset(offset) - offsetMap.delete(event.uuid!) } await heartbeat() await commitOffsetsIfNecessary() this.pluginsServer.statsd?.timing('kafka_queue.single_ingestion', singleIngestionTimer) } - for (const removedOffset of offsetMap.values()) { - resolveOffset(removedOffset) - await heartbeat() - await commitOffsetsIfNecessary() - } this.pluginsServer.statsd?.timing('kafka_queue.each_batch', batchProcessingTimer) } @@ -98,9 +80,7 @@ export class KafkaQueue implements Queue { await this.consumer.subscribe({ topic: KAFKA_EVENTS_INGESTION_HANDOFF }) // 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 - // The issue is right now we'd miss some messages and not resolve them as processEventBatch COMPETELY - // discards some events, leaving us with no kafka_offset to resolve when in fact it should be resolved. + eachBatchAutoResolve: true, autoCommitInterval: 500, // autocommit every 500 ms… autoCommitThreshold: 1000, // …or every 1000 messages, whichever is sooner eachBatch: this.eachBatch.bind(this), diff --git a/src/plugins.ts b/src/plugins.ts index 95ffb91e..b80976e8 100644 --- a/src/plugins.ts +++ b/src/plugins.ts @@ -264,7 +264,7 @@ export async function runPluginsOnBatch(server: PluginsServer, batch: PluginEven allReturnedEvents = allReturnedEvents.concat(returnedEvents) } - return allReturnedEvents + return allReturnedEvents.filter(Boolean) } export async function runPluginTask(server: PluginsServer, taskName: string, pluginConfigId: number): Promise { From 1515702597cd3b099532f08e59b9cfef9f43d873 Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Wed, 3 Feb 2021 21:53:39 +0100 Subject: [PATCH 4/9] sort events according to offsets --- src/ingestion/kafka-queue.ts | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/src/ingestion/kafka-queue.ts b/src/ingestion/kafka-queue.ts index 99dd3e4f..4ec4f908 100644 --- a/src/ingestion/kafka-queue.ts +++ b/src/ingestion/kafka-queue.ts @@ -37,9 +37,14 @@ export class KafkaQueue implements Queue { isStale, }: EachBatchPayload): Promise { const batchProcessingTimer = new Date() - const pluginEvents = batch.messages.map((message) => { + + const uuidOrder = new Map() + const uuidOffset = new Map() + const pluginEvents = batch.messages.map((message, index) => { const { data: dataStr, ...rawEvent } = JSON.parse(message.value!.toString()) const event = { ...rawEvent, ...JSON.parse(dataStr) } + uuidOrder.set(event.uuid, index) + uuidOffset.set(event.uuid, message.offset) return { ...event, site_url: event.site_url || null, @@ -48,6 +53,9 @@ export class KafkaQueue implements Queue { }) const processedEvents = await this.processEventBatch(pluginEvents) + processedEvents.sort( + (a, b) => (uuidOrder.get(a.uuid!) || pluginEvents.length) - (uuidOrder.get(b.uuid!) || pluginEvents.length) + ) for (const event of processedEvents) { if (!isRunning()) { @@ -60,7 +68,7 @@ export class KafkaQueue implements Queue { } const singleIngestionTimer = new Date() await this.saveEvent(event) - const offset = event?.uuid ? offsetMap.get(event.uuid) : null + const offset = event?.uuid ? uuidOffset.get(event.uuid) : null if (offset) { resolveOffset(offset) } From bc7555f6e83f02d25a20d92edbc88f5aece472f9 Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Wed, 3 Feb 2021 21:56:49 +0100 Subject: [PATCH 5/9] add sort info --- src/ingestion/kafka-queue.ts | 3 +++ 1 file changed, 3 insertions(+) diff --git a/src/ingestion/kafka-queue.ts b/src/ingestion/kafka-queue.ts index 4ec4f908..d3580689 100644 --- a/src/ingestion/kafka-queue.ts +++ b/src/ingestion/kafka-queue.ts @@ -53,6 +53,9 @@ export class KafkaQueue implements Queue { }) const processedEvents = await this.processEventBatch(pluginEvents) + + // Sort in the original order that the events came in, putting any randomly added events to the end. + // This is so we would resolve the correct kafka offsets in order. processedEvents.sort( (a, b) => (uuidOrder.get(a.uuid!) || pluginEvents.length) - (uuidOrder.get(b.uuid!) || pluginEvents.length) ) From 12ba2c63fdb6bfa0322ea87fcbbf27c0a67346c3 Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Wed, 3 Feb 2021 22:00:44 +0100 Subject: [PATCH 6/9] simplify offset line --- src/ingestion/kafka-queue.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/ingestion/kafka-queue.ts b/src/ingestion/kafka-queue.ts index d3580689..990e5171 100644 --- a/src/ingestion/kafka-queue.ts +++ b/src/ingestion/kafka-queue.ts @@ -71,7 +71,7 @@ export class KafkaQueue implements Queue { } const singleIngestionTimer = new Date() await this.saveEvent(event) - const offset = event?.uuid ? uuidOffset.get(event.uuid) : null + const offset = uuidOffset.get(event.uuid!) if (offset) { resolveOffset(offset) } From 3fd51b87d60550bef6ec307c51c10fd457771fa5 Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Wed, 3 Feb 2021 22:03:29 +0100 Subject: [PATCH 7/9] add comment --- src/ingestion/kafka-queue.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/ingestion/kafka-queue.ts b/src/ingestion/kafka-queue.ts index 990e5171..fd33742b 100644 --- a/src/ingestion/kafka-queue.ts +++ b/src/ingestion/kafka-queue.ts @@ -91,7 +91,7 @@ export class KafkaQueue implements Queue { await this.consumer.subscribe({ topic: KAFKA_EVENTS_INGESTION_HANDOFF }) // KafkaJS batching: https://kafka.js.org/docs/consuming#a-name-each-batch-a-eachbatch await this.consumer.run({ - eachBatchAutoResolve: true, + eachBatchAutoResolve: true, // commit the last offset of the batch if eachBatch doesn't throw an error autoCommitInterval: 500, // autocommit every 500 ms… autoCommitThreshold: 1000, // …or every 1000 messages, whichever is sooner eachBatch: this.eachBatch.bind(this), From b897fbd87e3175117ee78992fc4a52ef6ff177ce Mon Sep 17 00:00:00 2001 From: Michael Matloka Date: Thu, 4 Feb 2021 03:29:35 +0100 Subject: [PATCH 8/9] Type pluginEvents as PluginEvent[] --- src/ingestion/kafka-queue.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/ingestion/kafka-queue.ts b/src/ingestion/kafka-queue.ts index fd33742b..5804a0fe 100644 --- a/src/ingestion/kafka-queue.ts +++ b/src/ingestion/kafka-queue.ts @@ -40,7 +40,7 @@ export class KafkaQueue implements Queue { const uuidOrder = new Map() const uuidOffset = new Map() - const pluginEvents = batch.messages.map((message, index) => { + const pluginEvents: PluginEvent[] = batch.messages.map((message, index) => { const { data: dataStr, ...rawEvent } = JSON.parse(message.value!.toString()) const event = { ...rawEvent, ...JSON.parse(dataStr) } uuidOrder.set(event.uuid, index) From 5d6d1b89cf2308e90b38c31bd3a0fc01f04daf32 Mon Sep 17 00:00:00 2001 From: Michael Matloka Date: Thu, 4 Feb 2021 03:42:28 +0100 Subject: [PATCH 9/9] Commit last offset of batch manually --- src/ingestion/kafka-queue.ts | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/src/ingestion/kafka-queue.ts b/src/ingestion/kafka-queue.ts index 5804a0fe..8e8273d3 100644 --- a/src/ingestion/kafka-queue.ts +++ b/src/ingestion/kafka-queue.ts @@ -32,7 +32,6 @@ export class KafkaQueue implements Queue { resolveOffset, heartbeat, commitOffsetsIfNecessary, - uncommittedOffsets, isRunning, isStale, }: EachBatchPayload): Promise { @@ -80,6 +79,8 @@ export class KafkaQueue implements Queue { this.pluginsServer.statsd?.timing('kafka_queue.single_ingestion', singleIngestionTimer) } this.pluginsServer.statsd?.timing('kafka_queue.each_batch', batchProcessingTimer) + resolveOffset(batch.lastOffset()) + await commitOffsetsIfNecessary() } async start(): Promise { @@ -91,7 +92,7 @@ export class KafkaQueue implements Queue { await this.consumer.subscribe({ topic: KAFKA_EVENTS_INGESTION_HANDOFF }) // KafkaJS batching: https://kafka.js.org/docs/consuming#a-name-each-batch-a-eachbatch await this.consumer.run({ - eachBatchAutoResolve: true, // commit the last offset of the batch if eachBatch doesn't throw an error + eachBatchAutoResolve: false, // we are resolving the last offset of the batch more deliberately autoCommitInterval: 500, // autocommit every 500 ms… autoCommitThreshold: 1000, // …or every 1000 messages, whichever is sooner eachBatch: this.eachBatch.bind(this),