Optimize Converters.topicPartitionsSetF - #608
Optimize Converters.topicPartitionsSetF#608nviliunov-evolution-throwaway wants to merge 1 commit into
Conversation
|
📝 WalkthroughWalkthroughThe PR updates ChangesConversion and partition updates
Estimated code review effort: 3 (Moderate) | ~20 minutes 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Prompt for all review comments with 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.
Inline comments:
In `@skafka/src/main/scala/com/evolutiongaming/skafka/Converters.scala`:
- Line 19: Update the conversion method that currently loops through elements
with asScala[Try] to convert each element using asScala[F] and combine the
results through the existing ApplicativeThrowable[F] instance, preserving
accumulated errors instead of failing at the first invalid partition. If
accumulation cannot be supported, narrow the method contract to fail-fast
effects; add a regression test using an accumulating F instance.
- Around line 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.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: 783123df-838e-4357-a20d-fbda05a6f53d
📒 Files selected for processing (2)
skafka/src/main/scala/com/evolutiongaming/skafka/Converters.scalaskafka/src/main/scala/com/evolutiongaming/skafka/Partition.scala
| def topicPartitionsSetF[F[_]: ApplicativeThrowable](setJ: SetJ[TopicPartitionJ]): F[Set[TopicPartition]] = | ||
| 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() | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
gh run view 31409780871 --job 93524765345 --log-failedRepository: 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.scalaRepository: 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 HEADRepository: 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.scalaRepository: 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
There was a problem hiding this comment.
flaky test, green now
There was a problem hiding this comment.
@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.
Benchmark analysis by AI
|
| r <- setJ.asScala.toList.traverse { _.asScala[F] } | ||
| } yield r.toSet | ||
| } | ||
| def topicPartitionsSetF[F[_]: ApplicativeThrowable](setJ: SetJ[TopicPartitionJ]): F[Set[TopicPartition]] = |
There was a problem hiding this comment.
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?
There was a problem hiding this comment.
I agree, the current approach to labeling metrics is very bizzare
3 participants
Summary by CodeRabbit
Bug Fixes
Refactor