Skip to content

Optimize Converters.topicPartitionsSetF #608

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

Open
nviliunov-evolution-throwaway wants to merge 1 commit into
base: master
Choose a base branch
from
+19 −11
Open

Optimize Converters.topicPartitionsSetF #608

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
22 changes: 14 additions & 8 deletions skafka/src/main/scala/com/evolutiongaming/skafka/Converters.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,10 +2,9 @@ package com.evolutiongaming.skafka

import java.lang.Long as LongJ
import java.time.Duration as DurationJ
import java.util.{Optional, Collection as CollectionJ, Map as MapJ, Set as SetJ, List as ListJ}

import java.util.{Optional, Collection as CollectionJ, List as ListJ, Map as MapJ, Set as SetJ}
import cats.Monad
import cats.data.{NonEmptyList as Nel, NonEmptySet as Nes, NonEmptyMap as Nem}
import cats.data.{NonEmptyList as Nel, NonEmptyMap as Nem, NonEmptySet as Nes}
import cats.syntax.all.*
import com.evolutiongaming.catshelper.CatsHelper.*
import com.evolutiongaming.catshelper.{ApplicativeThrowable, FromTry, MonadThrowable, ToTry}
Expand All @@ -17,6 +16,7 @@ import org.apache.kafka.common.{PartitionInfo as PartitionInfoJ, TopicPartition
import scala.compat.java8.DurationConverters
import scala.concurrent.duration.FiniteDuration
import scala.jdk.CollectionConverters.*
import scala.util.{Failure, Success, Try}
Comment thread
coderabbitai[bot] marked this conversation as resolved.

object Converters {

Expand Down Expand Up @@ -233,9 +233,15 @@ object Converters {
mapJ.asScalaMap(_.pure[F], partitionsInfoListF[F])
}

def topicPartitionsSetF[F[_]: ApplicativeThrowable](setJ: SetJ[TopicPartitionJ]): F[Set[TopicPartition]] = {
for {
r <- setJ.asScala.toList.traverse { _.asScala[F] }
} yield r.toSet
}
def topicPartitionsSetF[F[_]: ApplicativeThrowable](setJ: SetJ[TopicPartitionJ]): F[Set[TopicPartition]] =

@mr-git mr-git Aug 11, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

I was thinking about the soundness of the caller of this API...

Very possibly this comment should be a dedicated issue! cc: @Z1kkurat

IIUC, we call it after every poll and we pass here full list of assigned topic-partitions. I wonder, if metrics are tracking "sound" data - should we report any metric for topic-partitions, which didn't return any data? Maybe we should convert only the set of topic-partitions, which delivered at least one message in last poll?

@nviliunov-evolution-throwaway nviliunov-evolution-throwaway Aug 11, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

I agree, the current approach to labeling metrics is very bizzare

ApplicativeThrowable[F].catchNonFatal {
val builder = Set.newBuilder[TopicPartition]
setJ.forEach { tpj =>
tpj.asScala[Try] match {
case Failure(exception) => throw exception
case Success(value) => builder.addOne(value)
}
}
builder.result()
}
Comment on lines +236 to +246

@coderabbitai coderabbitai Bot Aug 10, 2026 •
edited
Loading

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

🩺 Stability & Availability | 🟠 Major | ⚡ Quick win

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
set -euo pipefail
gh run view 31409780871 --job 93524765345 --log-failed

Repository: evolution-gaming/skafka

Length of output: 50379

🏁 Script executed:

set -euo pipefail
log=/tmp/coderabbit-shell-logs/shell-output-7Vo8pu
printf '%s\n' '--- failure and conclusion markers ---'
rg -n -i 'error|failed|failure|aborted|exit code|process completed|all tests passed|success|cancelled|timed out' "$log" | tail -n 120
printf '%s\n' '--- final log lines ---'
tail -n 160 "$log"

Repository: evolution-gaming/skafka

Length of output: 46034

🏁 Script executed:

set -euo pipefail
printf '%s\n' '--- changed files ---'
git diff --stat
printf '%s\n' '--- Converters.scala implementation and callers ---'
rg -n -C 8 'topicPartitionsSetF|SetJ\[TopicPartitionJ\]' skafka/src/main/scala skafka/src/test/scala
printf '%s\n' '--- failing test context ---'
sed -n '120,155p' skafka/src/test/scala/com/evolutiongaming/skafka/ProducerConsumerSpec.scala
sed -n '415,440p' skafka/src/test/scala/com/evolutiongaming/skafka/ProducerConsumerSpec.scala
printf '%s\n' '--- Converters.scala diff ---'
git diff -- skafka/src/main/scala/com/evolutiongaming/skafka/Converters.scala

Repository: evolution-gaming/skafka

Length of output: 41603

🏁 Script executed:

set -euo pipefail
printf '%s\n' '--- repository state ---'
git status --short
git rev-parse --short HEAD
printf '%s\n' '--- ProducerConsumerSpec location ---'
fd -t f 'ProducerConsumerSpec.scala' .
printf '%s\n' '--- test declaration and timeout helper ---'
file=$(fd -t f 'ProducerConsumerSpec.scala' . | head -n 1)
rg -n -C 12 'rebalance listener correctness|consumer\.commit|33 seconds|def timeout|TimeoutException' "$file"
printf '%s\n' '--- commit files for the reviewed path ---'
git diff --name-status
git show --stat --oneline --decorate HEAD

Repository: evolution-gaming/skafka

Length of output: 7120

🏁 Script executed:

set -euo pipefail
git show --format=fuller --no-ext-diff -- skafka/src/main/scala/com/evolutiongaming/skafka/Converters.scala skafka/src/main/scala/com/evolutiongaming/skafka/Partition.scala
printf '%s\n' '--- commit-path call graph ---'
rg -n -C 10 'def commit|commit\(|onPartitionsRevoked|RebalanceListener1WithConsumer|topicPartitionsSetF' \
  skafka/src/main/scala/com/evolutiongaming/skafka/consumer \
  tests/src/test/scala/com/evolutiongaming/skafka/ProducerConsumerSpec.scala

Repository: evolution-gaming/skafka

Length of output: 50380

Resolve the Scala 3.3.8 test failure before merging.

tests/src/test/scala/com/evolutiongaming/skafka/ProducerConsumerSpec.scala fails rebalance listener correctness - consumer.commit with a 33-second TimeoutException.

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@skafka/src/main/scala/com/evolutiongaming/skafka/Converters.scala` around
lines 236 - 246, Fix the Scala 3.3.8 timeout in the rebalance listener’s
consumer.commit path by tracing the failure through
Converters.topicPartitionsSetF and the related commit flow. Ensure Java topic
partitions are converted reliably into the complete Scala Set and that
conversion failures propagate through ApplicativeThrowable without disrupting
the consumer callback; verify ProducerConsumerSpec’s “rebalance listener
correctness - consumer.commit” test passes.

Source: MCP tools

@nviliunov-evolution-throwaway nviliunov-evolution-throwaway Aug 10, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

flaky test, green now

@coderabbitai coderabbitai Bot Aug 10, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

@nviliunov-evolution-throwaway, acknowledged. The failing test was flaky and is green now. I will treat the reported timeout as intermittent rather than as confirmed regression from this PR.

You are interacting with an AI system.

}
8 changes: 5 additions & 3 deletions skafka/src/main/scala/com/evolutiongaming/skafka/Partition.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,9 +14,11 @@ sealed abstract case class Partition(value: Int) {

object Partition {

val min: Partition = new Partition(0) {}
private class Impl(value: Int) extends Partition(value)

val max: Partition = new Partition(Int.MaxValue) {}
val min: Partition = new Impl(0)

val max: Partition = new Impl(Int.MaxValue)

implicit val showPartition: Show[Partition] = Show.fromToString

Expand All @@ -36,7 +38,7 @@ object Partition {
} else if (value == max.value) {
max.pure[F]
} else {
new Partition(value) {}.pure[F]
(new Impl(value): Partition).pure[F]
}
}

Expand Down