Skip to content

fix(flow): grupo de consumidor próprio na fila de gatilhos de jornada do ClickHouse (CRM-777) #128

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
gomessguii merged 1 commit into from
Oct 8, 2026
Merged

fix(flow): grupo de consumidor próprio na fila de gatilhos de jornada do ClickHouse (CRM-777) #128

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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: 2 additions & 0 deletions src/modules/processing/clickhouse/clickhouse.service.contact-events-broker.spec.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@ type ServiceInternals = {
tableName: string,
expectedBrokers: string,
dependentViews?: string[],
expectedGroup?: string,
): Promise<boolean>;
extractKafkaBrokers(createTableQuery: string): string | null;
query: jest.Mock;
Expand Down Expand Up @@ -147,6 +148,7 @@ describe('ClickHouseService — contact-events broker guard (EVO-1925)', () => {
'contact_events_kafka_queue',
'evo-campaign-kafka:29092',
['contact_events_kafka_mv'],
'evo-campaign-consumers-clickhouse',
);

// Guard must run before the CREATE TABLE command for the Kafka queue.
Expand Down
108 changes: 108 additions & 0 deletions src/modules/processing/clickhouse/clickhouse.service.spec.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,8 +28,11 @@ type ServiceInternals = {
tableName: string,
expectedBrokers: string,
dependentViews?: string[],
expectedGroup?: string,
): Promise<boolean>;
extractKafkaBrokers(createTableQuery: string): string | null;
extractKafkaGroup(createTableQuery: string): string | null;
createJourneyTriggerQueue(databaseName: string): Promise<void>;
query: jest.Mock;
command: jest.Mock;
logger: { log: jest.Mock; warn: jest.Mock; error: jest.Mock; debug: jest.Mock };
Expand All @@ -38,6 +41,10 @@ type ServiceInternals = {
describe('ClickHouseService — journey-trigger broker guard (EVO-1893)', () => {
let service: ClickHouseService;
let internals: ServiceInternals;
const commandQueries = () =>
(internals.command.mock.calls as [{ query: string }][]).map(
([call]) => call.query,
);

beforeEach(() => {
service = new ClickHouseService();
Expand Down Expand Up @@ -65,6 +72,28 @@ describe('ClickHouseService — journey-trigger broker guard (EVO-1893)', () =>
internals.extractKafkaBrokers('ENGINE = MergeTree() ORDER BY x'),
).toBeNull();
});

it('extracts the consumer group from a Kafka Engine DDL', () => {
const ddl =
"ENGINE = Kafka('kafka-external:9094', 'journey-triggers', 'temporal-workers', 'JSONEachRow') SETTINGS kafka_num_consumers = 2";
expect(internals.extractKafkaGroup(ddl)).toBe('temporal-workers');
expect(
internals.extractKafkaGroup('ENGINE = MergeTree() ORDER BY x'),
).toBeNull();
});

it('reads broker and group from a DDL split across lines', () => {
const ddl = [
'ENGINE = Kafka(',
" 'kafka-external:9094',",
" 'journey-triggers',",
" 'temporal-workers',",
" 'JSONEachRow'",
')',
].join('\n');
expect(internals.extractKafkaBrokers(ddl)).toBe('kafka-external:9094');
expect(internals.extractKafkaGroup(ddl)).toBe('temporal-workers');
});
});

describe('ensureKafkaEngineBroker', () => {
Expand Down Expand Up @@ -132,6 +161,57 @@ describe('ClickHouseService — journey-trigger broker guard (EVO-1893)', () =>
);
});

it('drops the table when only the consumer group is stale', async () => {
internals.query.mockResolvedValue([
{
engine: 'Kafka',
create:
"ENGINE = Kafka('evo-campaign-kafka:29092', 'journey-triggers', 'temporal-workers', 'JSONEachRow')",
},
]);

const dropped = await internals.ensureKafkaEngineBroker(
'evo_campaign',
'journey_trigger_kafka_queue',
'evo-campaign-kafka:29092',
['events_to_journey_triggers_mv'],
'evo-campaign-consumers-journey-triggers-clickhouse',
);

expect(dropped).toBe(true);
const commands = commandQueries();
expect(commands[0]).toContain(
'DROP VIEW IF EXISTS evo_campaign.events_to_journey_triggers_mv',
);
expect(commands[1]).toContain(
'DROP TABLE IF EXISTS evo_campaign.journey_trigger_kafka_queue',
);
expect(internals.logger.warn).toHaveBeenCalledWith(
expect.stringContaining("uses consumer group 'temporal-workers'"),
);
});

it('keeps the table when broker and group both match', async () => {
internals.query.mockResolvedValue([
{
engine: 'Kafka',
create:
"ENGINE = Kafka('evo-campaign-kafka:29092', 'journey-triggers', 'evo-campaign-consumers-journey-triggers-clickhouse', 'JSONEachRow')",
},
]);

const dropped = await internals.ensureKafkaEngineBroker(
'evo_campaign',
'journey_trigger_kafka_queue',
'evo-campaign-kafka:29092',
['events_to_journey_triggers_mv'],
'evo-campaign-consumers-journey-triggers-clickhouse',
);

expect(dropped).toBe(false);
expect(internals.command).not.toHaveBeenCalled();
});

it('leaves non-Kafka tables untouched', async () => {
internals.query.mockResolvedValue([
{ engine: 'MergeTree', create: 'ENGINE = MergeTree() ORDER BY x' },
Expand Down Expand Up @@ -162,4 +242,32 @@ describe('ClickHouseService — journey-trigger broker guard (EVO-1893)', () =>
expect(internals.logger.error).toHaveBeenCalled();
});
});

describe('createJourneyTriggerQueue wiring', () => {
it("binds the queue to ClickHouse's own consumer group, never the journey workers'", async () => {
const ensureSpy = jest
.spyOn(internals as any, 'ensureKafkaEngineBroker')
.mockResolvedValue(false);
internals.query.mockResolvedValue([]);

await internals.createJourneyTriggerQueue('evo_campaign');

expect(ensureSpy).toHaveBeenCalledWith(
'evo_campaign',
'journey_trigger_kafka_queue',
'evo-campaign-kafka:29092',
['events_to_journey_triggers_mv'],
'evo-campaign-consumers-journey-triggers-clickhouse',
);
const createQueue = commandQueries().find((q) =>
q.includes(
'CREATE TABLE IF NOT EXISTS evo_campaign.journey_trigger_kafka_queue',
),
);
expect(createQueue).toContain(
"Kafka('evo-campaign-kafka:29092', 'journey-triggers', 'evo-campaign-consumers-journey-triggers-clickhouse', 'JSONEachRow')",
);
expect(createQueue).not.toContain("'temporal-workers'");
});
});
});
51 changes: 40 additions & 11 deletions src/modules/processing/clickhouse/clickhouse.service.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -337,9 +337,9 @@ export class ClickHouseService implements OnModuleInit, OnModuleDestroy {
* broker string from the first boot. If that boot lacked KAFKA_BROKERS_INTERNAL
* the table is stuck on 'localhost:9092' forever — ClickHouse can't reach Kafka
* and the topic silently receives nothing (EVO-1893/EVO-1925). This reads the
* existing DDL, and if the broker diverges from `expectedBrokers`, drops the
* dependent materialized views and the Kafka table so the caller can recreate
* them with the right broker.
* existing DDL, and if the broker diverges from `expectedBrokers` (or the
* consumer group from `expectedGroup`, when given), drops the dependent
* materialized views and the Kafka table so the caller can recreate them.
*
* @returns true if the table was dropped (stale), false if absent or already correct.
*/
Expand All @@ -348,6 +348,7 @@ export class ClickHouseService implements OnModuleInit, OnModuleDestroy {
tableName: string,
expectedBrokers: string,
dependentViews: string[] = [],
expectedGroup?: string,
): Promise<boolean> {
try {
const rows = await this.query<{ engine: string; create: string }>({
Expand All @@ -372,19 +373,32 @@ export class ClickHouseService implements OnModuleInit, OnModuleDestroy {
}

const currentBrokers = this.extractKafkaBrokers(existing.create);
const currentGroup = this.extractKafkaGroup(existing.create);
const staleBroker = currentBrokers !== expectedBrokers;
const staleGroup =
expectedGroup !== undefined && currentGroup !== expectedGroup;

if (currentBrokers === expectedBrokers) {
if (!staleBroker && !staleGroup) {
this.logger.log(
`Kafka Engine table '${databaseName}.${tableName}' already points at the configured broker ('${expectedBrokers}')`,
`Kafka Engine table '${databaseName}.${tableName}' already points at the configured broker ('${expectedBrokers}')` +
(expectedGroup ? ` and group ('${expectedGroup}')` : ''),
);
return false;
}

this.logger.warn(
`⚠️ Kafka Engine table '${databaseName}.${tableName}' is bound to a stale broker ` +
`('${currentBrokers ?? 'unknown'}') but the configured broker is '${expectedBrokers}'. ` +
`Recreating it so the pipeline reaches Kafka (EVO-1893/EVO-1925).`,
);
if (staleBroker) {
this.logger.warn(
`⚠️ Kafka Engine table '${databaseName}.${tableName}' is bound to a stale broker ` +
`('${currentBrokers ?? 'unknown'}') but the configured broker is '${expectedBrokers}'. ` +
`Recreating it so the pipeline reaches Kafka (EVO-1893/EVO-1925).`,
);
}
if (staleGroup) {
this.logger.warn(
`⚠️ Kafka Engine table '${databaseName}.${tableName}' uses consumer group ` +
`'${currentGroup ?? 'unknown'}' but the configured group is '${expectedGroup}'. Recreating it.`,
);
}

// Drop dependent materialized views first (they block dropping the source).
for (const view of dependentViews) {
Expand Down Expand Up @@ -421,6 +435,16 @@ export class ClickHouseService implements OnModuleInit, OnModuleDestroy {
return match ? match[1] : null;
}

/**
* Extract the consumer group (third argument) from a Kafka Engine DDL.
*/
private extractKafkaGroup(createTableQuery: string): string | null {
const match = createTableQuery.match(
/Kafka\(\s*'[^']*'\s*,\s*'[^']*'\s*,\s*'([^']*)'/i,
);
return match ? match[1] : null;
}

/**
* Log the broker the contact-events Kafka Engine table is actually bound to.
* The Kafka Engine consumes/produces in the background and does not raise to
Expand Down Expand Up @@ -486,6 +510,7 @@ export class ClickHouseService implements OnModuleInit, OnModuleDestroy {
// The MV reads from the Kafka queue and writes into the main table; it
// must be dropped before the underlying Kafka table can be dropped.
[`${tableName}_kafka_mv`],
kafkaGroupId,
);

// 1. Create Kafka queue table
Expand Down Expand Up @@ -930,7 +955,10 @@ export class ClickHouseService implements OnModuleInit, OnModuleDestroy {

const kafkaBrokers = this.config.kafka?.brokersInternal || 'kafka:29092';
const kafkaTopic = 'journey-triggers';
const kafkaGroupId = 'temporal-workers';
// ClickHouse only produces into this topic, but the table still names a
// consumer group: never the journey workers' own 'temporal-workers', or a
// consumer started on it would split their partitions and drop triggers.
const kafkaGroupId = `${this.config.kafka?.groupId || 'evo-campaign-consumers'}-journey-triggers-clickhouse`;

this.logger.log(
`Using Kafka brokers for Journey Triggers: ${kafkaBrokers}`,
Expand All @@ -949,6 +977,7 @@ export class ClickHouseService implements OnModuleInit, OnModuleDestroy {
// The MV reads from contact_events and writes into the queue table; it
// must be dropped before the underlying Kafka table can be dropped.
['events_to_journey_triggers_mv'],
kafkaGroupId,
);

// 1. Create Kafka table for journey triggers
Expand Down
Loading