SharedKafkaConsumer.unSubscribe() was missing the timeoutMs param…

…eter, which was overwritten as a merge conflict in #1308. 🫨
linkedin · Dec 20, 2024 · e0a23f9 · e0a23f9
1 parent ed4776f
commit e0a23f9
Showing 1 changed file with 1 addition and 1 deletion.
diff --git a/...-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/KafkaConsumerService.java b/...-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/KafkaConsumerService.java
@@ -262,7 +262,7 @@ public void unSubscribe(PubSubTopic versionTopic, PubSubTopicPartition pubSubTop
        * setting data receiver and unsubscribing concurrently for the same topic partition on a shared consumer.
        */
       try (AutoCloseableLock ignored = AutoCloseableLock.of(consumerToLocks.get(consumer))) {
-        consumer.unSubscribe(pubSubTopicPartition);
+        consumer.unSubscribe(pubSubTopicPartition, timeoutMs);
         removeTopicPartitionFromConsumptionTask(consumer, pubSubTopicPartition);
       }
       versionTopicToTopicPartitionToConsumer.compute(versionTopic, (k, topicPartitionToConsumerMap) -> {