dajac commented on code in PR #13112:
URL: https://github.com/apache/kafka/pull/13112#discussion_r1073960778
##########
core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala:
##########
@@ -810,19 +810,22 @@ class GroupMetadataManager(brokerId: Int,
*/
private def maybeUpdateCoordinatorEpoch(
partitionId: Int,
- epochOpt: Option[Int]
+ epochOpt: OptionalInt
): Boolean = {
val updatedEpoch = epochForPartitionId.compute(partitionId, (_,
currentEpoch) => {
if (currentEpoch == null) {
- epochOpt.map(Int.box).orNull
+ if (epochOpt.isPresent) epochOpt.getAsInt
+ else null
} else {
- epochOpt match {
- case Some(epoch) if epoch > currentEpoch => epoch
- case _ => currentEpoch
- }
+ if (epochOpt.isPresent && epochOpt.getAsInt > currentEpoch)
epochOpt.getAsInt
+ else currentEpoch
}
})
- epochOpt.forall(_ == updatedEpoch)
+ if(epochOpt.isPresent) {
Review Comment:
> I also wonder if there is a cleaner way to write the above methods with
the OptionalInt. But perhaps not.
This is the best I came up with.
> I see the options here are limited. Did you consider using java optionals
here too? I suppose those are a bit more heavyweight and maybe still don't
provide much better methods.
I don't understand your point here. We already use java optionals here now.
##########
core/src/main/scala/kafka/coordinator/group/GroupCoordinatorAdapter.scala:
##########
@@ -511,4 +532,53 @@ class GroupCoordinatorAdapter(
future
}
+
+ override def partitionFor(groupId: String): Int = {
+ coordinator.partitionFor(groupId)
+ }
+
+ override def onTransactionCompleted(
Review Comment:
I found the old name a bit weird. I have also tried to name all these
"callbacks" similarly. They all start with `on` now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
