Skip to content
This repository was archived by the owner on Nov 4, 2021. It is now read-only.

Remove "_handoff" from Kafka topic #127

Merged
mariusandra merged 3 commits into from
Feb 4, 2021
Merged

Remove "_handoff" from Kafka topic #127

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
b2c22e9
remove _HANDOFF from topic
mariusandra Feb 4, 2021
3b80c73
add plugin_ to plugin server ingestion topic
mariusandra Feb 4, 2021
01fa5ae
Merge branch 'ingestion-save' into remove-handoff
mariusandra Feb 4, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion src/ingestion/topics.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,4 +3,4 @@ export const KAFKA_PERSON = 'clickhouse_person'
export const KAFKA_PERSON_UNIQUE_ID = 'clickhouse_person_unique_id'
export const KAFKA_SESSION_RECORDING_EVENTS = 'clickhouse_session_recording_events'
export const KAFKA_EVENTS_WAL = 'events_write_ahead_log'
export const KAFKA_EVENTS_INGESTION_HANDOFF = 'events_ingestion_handoff'
export const KAFKA_EVENTS_PLUGIN_INGESTION = 'events_plugin_ingestion'
4 changes: 2 additions & 2 deletions src/server.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +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'
import { KAFKA_EVENTS_PLUGIN_INGESTION, KAFKA_EVENTS_WAL } from './ingestion/topics'

export async function createServer(
config: Partial<PluginsServerConfig> = {},
Expand Down Expand Up @@ -86,7 +86,7 @@ export async function createServer(
if (!serverConfig.KAFKA_CONSUMPTION_TOPIC) {
// When ingesting events, listen to the "INGESTION_HANDOFF" topic, otherwise listen to the "WAL" and discard
serverConfig.KAFKA_CONSUMPTION_TOPIC = serverConfig.PLUGIN_SERVER_INGESTION
? KAFKA_EVENTS_INGESTION_HANDOFF
? KAFKA_EVENTS_PLUGIN_INGESTION
: KAFKA_EVENTS_WAL
}

Expand Down
4 changes: 2 additions & 2 deletions tests/clickhouse/e2e.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +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'
import { KAFKA_EVENTS_PLUGIN_INGESTION } from '../../src/ingestion/topics'

jest.setTimeout(60000) // 60 sec timeout

Expand All @@ -18,7 +18,7 @@ const extraServerConfig: Partial<PluginsServerConfig> = {
KAFKA_HOSTS: process.env.KAFKA_HOSTS || 'kafka:9092',
WORKER_CONCURRENCY: 2,
PLUGIN_SERVER_INGESTION: true,
KAFKA_CONSUMPTION_TOPIC: KAFKA_EVENTS_INGESTION_HANDOFF,
KAFKA_CONSUMPTION_TOPIC: KAFKA_EVENTS_PLUGIN_INGESTION,
LOG_LEVEL: LogLevel.Log,
}

Expand Down
4 changes: 2 additions & 2 deletions tests/clickhouse/process-event.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,15 +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'
import { KAFKA_EVENTS_PLUGIN_INGESTION } from '../../src/ingestion/topics'

jest.setTimeout(180_000) // 3 minute timeout

const extraServerConfig: Partial<PluginsServerConfig> = {
KAFKA_ENABLED: true,
KAFKA_HOSTS: process.env.KAFKA_HOSTS || 'kafka:9092',
PLUGIN_SERVER_INGESTION: true,
KAFKA_CONSUMPTION_TOPIC: KAFKA_EVENTS_INGESTION_HANDOFF,
KAFKA_CONSUMPTION_TOPIC: KAFKA_EVENTS_PLUGIN_INGESTION,
}

describe('process event (clickhouse)', () => {
Expand Down
6 changes: 3 additions & 3 deletions tests/helpers/kafka.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@ import { delay, UUIDT } from '../../src/utils'
import { defaultConfig, overrideWithEnv } from '../../src/config'
import {
KAFKA_EVENTS,
KAFKA_EVENTS_INGESTION_HANDOFF,
KAFKA_EVENTS_PLUGIN_INGESTION,
KAFKA_EVENTS_WAL,
KAFKA_PERSON,
KAFKA_PERSON_UNIQUE_ID,
Expand All @@ -28,7 +28,7 @@ export async function resetKafka(extraServerConfig: Partial<PluginsServerConfig>

await createTopics(kafka, [
KAFKA_EVENTS,
KAFKA_EVENTS_INGESTION_HANDOFF,
KAFKA_EVENTS_PLUGIN_INGESTION,
KAFKA_EVENTS_WAL,
KAFKA_SESSION_RECORDING_EVENTS,
KAFKA_PERSON,
Expand All @@ -50,7 +50,7 @@ export async function resetKafka(extraServerConfig: Partial<PluginsServerConfig>
await producer.connect()
console.info('subscribing consumer')

await consumer.subscribe({ topic: KAFKA_EVENTS_INGESTION_HANDOFF })
await consumer.subscribe({ topic: KAFKA_EVENTS_PLUGIN_INGESTION })
console.info('running consumer')
await consumer.run({
eachMessage: async (payload) => {
Expand Down