Skip to content
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

[OTE-846] Bazooka sequential clear #2423

Merged
merged 3 commits into from
Oct 1, 2024
Merged

[OTE-846] Bazooka sequential clear #2423

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
b047742
enable orderId logging in print-block script
jerryfan01234 Sep 30, 2024
445735a
Change bazooka to clear topics sequentially
jerryfan01234 Oct 1, 2024
aa17e0b
Revert "enable orderId logging in print-block script"
jerryfan01234 Oct 1, 2024
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
19 changes: 11 additions & 8 deletions indexer/services/bazooka/src/index.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -264,14 +264,17 @@ async function partitionKafkaTopics(): Promise<void> {
async function clearKafkaTopics(
existingKafkaTopics: string[],
): Promise<void> {
await Promise.all(
_.map(KAFKA_TOPICS,
clearKafkaTopic.bind(null,
1,
config.CLEAR_KAFKA_TOPIC_RETRY_MS,
config.CLEAR_KAFKA_TOPIC_MAX_RETRIES,
existingKafkaTopics)),
);
// Concurrent calls to clear all topics caused the failure:
// TypeError: Cannot destructure property 'partitions' of 'high.pop(...)' as it is undefined.
for (const topic of KAFKA_TOPICS) {
await clearKafkaTopic(
1,
config.CLEAR_KAFKA_TOPIC_RETRY_MS,
config.CLEAR_KAFKA_TOPIC_MAX_RETRIES,
existingKafkaTopics,
topic,
);
Comment on lines +271 to +276
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue

Inconsistency in Parameter Defaults

In the call to clearKafkaTopic, the attempt parameter is explicitly set to 1, but the function clearKafkaTopic already defaults attempt to 1. Similarly, retryMs and maxRetries default to the same values in the function signature.

You can simplify the function call by relying on the default parameter values:

 await clearKafkaTopic(
-  1,
-  config.CLEAR_KAFKA_TOPIC_RETRY_MS,
-  config.CLEAR_KAFKA_TOPIC_MAX_RETRIES,
   existingKafkaTopics,
   topic,
 );

If you need to override the defaults, ensure that the arguments are necessary.

📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
1,
config.CLEAR_KAFKA_TOPIC_RETRY_MS,
config.CLEAR_KAFKA_TOPIC_MAX_RETRIES,
existingKafkaTopics,
topic,
);
await clearKafkaTopic(
existingKafkaTopics,
topic,
);

}
Comment on lines +269 to +277
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🛠️ Refactor suggestion

Potential Performance Impact of Sequential Topic Clearing

Changing from concurrent to sequential clearing of Kafka topics resolves the TypeError, but it may introduce performance bottlenecks, especially when dealing with a large number of topics. Consider implementing controlled concurrency to improve performance while avoiding the error.

You can use a concurrency control utility like p-limit to limit the number of concurrent operations. Here's how you might adjust the code:

+import pLimit from 'p-limit';

 async function clearKafkaTopics(
   existingKafkaTopics: string[],
 ): Promise<void> {
+  const limit = pLimit(5); // Adjust the concurrency level as appropriate
+
   // Concurrent calls to clear all topics caused the failure:
   // TypeError: Cannot destructure property 'partitions' of 'high.pop(...)' as it is undefined.
-  for (const topic of KAFKA_TOPICS) {
-    await clearKafkaTopic(
-      1,
-      config.CLEAR_KAFKA_TOPIC_RETRY_MS,
-      config.CLEAR_KAFKA_TOPIC_MAX_RETRIES,
-      existingKafkaTopics,
-      topic,
-    );
-  }
+  const clearTopicPromises = KAFKA_TOPICS.map((topic) =>
+    limit(() =>
+      clearKafkaTopic(
+        1,
+        config.CLEAR_KAFKA_TOPIC_RETRY_MS,
+        config.CLEAR_KAFKA_TOPIC_MAX_RETRIES,
+        existingKafkaTopics,
+        topic,
+      ),
+    ),
+  );
+  await Promise.all(clearTopicPromises);
 }

This approach balances performance with reliability by limiting the number of concurrent operations.

Committable suggestion was skipped due to low confidence.

}

export async function clearKafkaTopic(
Expand Down
Loading