Skip to content

Commit

Permalink
Add repartitioning step to bazooka (backport #2194) (#2203)
Browse files Browse the repository at this point in the history
Co-authored-by: roy-dydx <[email protected]>
  • Loading branch information
mergify[bot] and roy-dydx authored Sep 5, 2024
1 parent ac601db commit e1d3477
Showing 1 changed file with 25 additions and 1 deletion.
26 changes: 25 additions & 1 deletion indexer/services/bazooka/src/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@ const KAFKA_TOPICS_TO_PARTITIONS: { [key in KafkaTopics]: number } = {
[KafkaTopics.TO_ENDER]: 1,
[KafkaTopics.TO_VULCAN]: 60,
[KafkaTopics.TO_WEBSOCKETS_ORDERBOOKS]: 1,
[KafkaTopics.TO_WEBSOCKETS_SUBACCOUNTS]: 1,
[KafkaTopics.TO_WEBSOCKETS_SUBACCOUNTS]: 30,
[KafkaTopics.TO_WEBSOCKETS_TRADES]: 1,
[KafkaTopics.TO_WEBSOCKETS_MARKETS]: 1,
[KafkaTopics.TO_WEBSOCKETS_CANDLES]: 1,
Expand Down Expand Up @@ -182,6 +182,7 @@ async function maybeClearAndCreateKafkaTopics(

if (event.create_kafka_topics) {
await createKafkaTopics(existingKafkaTopics);
await partitionKafkaTopics();
}

if (event.clear_kafka_topics) {
Expand Down Expand Up @@ -233,6 +234,29 @@ async function createKafkaTopics(
});
}

async function partitionKafkaTopics(): Promise<void> {
for (const kafkaTopic of KAFKA_TOPICS) {
const topicMetadata: { topics: Array<ITopicMetadata> } = await admin.fetchTopicMetadata({
topics: [kafkaTopic],
});
if (topicMetadata.topics.length === 1) {
if (topicMetadata.topics[0].partitions.length !== KAFKA_TOPICS_TO_PARTITIONS[kafkaTopic]) {
await admin.createPartitions({
validateOnly: false,
topicPartitions: [{
topic: kafkaTopic,
count: KAFKA_TOPICS_TO_PARTITIONS[kafkaTopic],
}],
});
logger.info({
at: 'index#createKafka Topics',
message: `Successfully set topic ${kafkaTopic} to ${KAFKA_TOPICS_TO_PARTITIONS[kafkaTopic]} partitions`,
});
}
}
}
}

async function clearKafkaTopics(
existingKafkaTopics: string[],
): Promise<void> {
Expand Down

0 comments on commit e1d3477

Please sign in to comment.