Fix bug in clearing topics (backport #2202) (#2207)

Co-authored-by: roy-dydx <[email protected]>

indexer/services/bazooka/src/index.ts

@@ -234,11 +234,41 @@ async function createKafkaTopics(
   });
 }
 
+<<<<<<< HEAD
+=======
+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]) {
+        logger.info({
+          at: 'index#partitionKafkaTopics',
+          message: `Setting topic ${kafkaTopic} to ${KAFKA_TOPICS_TO_PARTITIONS[kafkaTopic]} partitions`,
+        });
+        await admin.createPartitions({
+          validateOnly: false,
+          topicPartitions: [{
+            topic: kafkaTopic,
+            count: KAFKA_TOPICS_TO_PARTITIONS[kafkaTopic],
+          }],
+        });
+        logger.info({
+          at: 'index#partitionKafkaTopics',
+          message: `Successfully set topic ${kafkaTopic} to ${KAFKA_TOPICS_TO_PARTITIONS[kafkaTopic]} partitions`,
+        });
+      }
+    }
+  }
+}
+
+>>>>>>> 0788d298 (Fix bug in clearing topics (#2202))
 async function clearKafkaTopics(
   existingKafkaTopics: string[],
 ): Promise<void> {
   await Promise.all(
-    _.map(KAFKA_TOPICS_TO_PARTITIONS,
+    _.map(KAFKA_TOPICS,
       clearKafkaTopic.bind(null,
         1,
         config.CLEAR_KAFKA_TOPIC_RETRY_MS,