Handle non-existent topics in partitionsFor

evolution-gaming · Nov 25, 2024 · a636a39 · a636a39
1 parent 04edf3e
commit a636a39
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 11 deletions.
diff --git a/skafka/src/main/scala/com/evolutiongaming/skafka/consumer/Consumer.scala b/skafka/src/main/scala/com/evolutiongaming/skafka/consumer/Consumer.scala
@@ -340,11 +340,11 @@ object Consumer {
         } yield result
       }
 
-      def partitions1(f: => ListJ[PartitionInfoJ]) = {
+      def partitions1(f: => Option[ListJ[PartitionInfoJ]]) = {
         for {
           result <- serialBlocking { f }
-          result <- partitionsInfoListF[F](result)
-        } yield result
+          result <- result.traverse(partitionsInfoListF[F])
+        } yield result.getOrElse(List.empty)
       }
 
       def topics1(f: => MapJ[Topic, ListJ[PartitionInfoJ]]) = {
@@ -506,11 +506,11 @@ object Consumer {
         }
 
         def partitions(topic: Topic) = {
-          partitions1 { consumer.partitionsFor(topic) }
+          partitions1 { Option(consumer.partitionsFor(topic)) }
         }
 
         def partitions(topic: Topic, timeout: FiniteDuration) = {
-          partitions1 { consumer.partitionsFor(topic, timeout.asJava) }
+          partitions1 { Option(consumer.partitionsFor(topic, timeout.asJava)) }
         }
 
         val topics = topics1 { consumer.listTopics() }

diff --git a/skafka/src/main/scala/com/evolutiongaming/skafka/consumer/RebalanceConsumer.scala b/skafka/src/main/scala/com/evolutiongaming/skafka/consumer/RebalanceConsumer.scala
@@ -5,6 +5,7 @@ import java.time.Instant
 import java.util.{Map => MapJ}
 
 import cats.data.{NonEmptyMap => Nem, NonEmptySet => Nes}
+import cats.implicits.toTraverseOps
 import com.evolutiongaming.skafka.Converters._
 import com.evolutiongaming.skafka.consumer.ConsumerConverters._
 import com.evolutiongaming.skafka._
@@ -217,15 +218,15 @@ object RebalanceConsumer {
 
       def partitionsFor(topic: Topic) =
         for {
-          a <- Try { c.partitionsFor(topic) }
-          a <- partitionsInfoListF[Try](a)
-        } yield a
+          a <- Try { Option(c.partitionsFor(topic)) }
+          a <- a.traverse(partitionsInfoListF[Try])
+        } yield a.getOrElse(List.empty)
 
       def partitionsFor(topic: Topic, timeout: FiniteDuration) =
         for {
-          a <- Try { c.partitionsFor(topic, timeout.asJava) }
-          a <- partitionsInfoListF[Try](a)
-        } yield a
+          a <- Try { Option(c.partitionsFor(topic, timeout.asJava)) }
+          a <- a.traverse(partitionsInfoListF[Try])
+        } yield a.getOrElse(List.empty)
 
       def paused() =
         for {