You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2021/10/25 15:40:10 UTC
[GitHub] [kafka] jlprat opened a new pull request #11432: KAFKA-13399 towards scala3
jlprat opened a new pull request #11432:
URL: https://github.com/apache/kafka/pull/11432
This PR takes all changes in https://github.com/apache/kafka/pull/11350 that are needed for a Scala 3 compatibility except those with an open bug report or already solved one.
The PR is separated in 5 commits, one for each type of change performed:
- Avoid Shadowing: Scala 3 compiler is not so friendly with shadowing and reports errors where Scala 2 wasn't.
- Not rely on automatic widening in numeric types: Scala 3 doesn't automatically convert `Short` to `Int` or `Int` to `Long` at will as in Scala 2. Changes in here help the typer by manually forcing the conversions.
- Remove redundant parenthesis: Scala 2 was more lax about calling with empty parenthesis a method without parenthesis. This became stricter in Scala 3.
- Explicit type declaration: Scala 3 changed a bit in the area of type inference and in some cases it picks the most general type, causing in our case some trouble as this is not public but package protected.
In other cases, the typer wasn't able to infer the proper one, forcing it to be manually set.
- Workaround for SAM conversion with overloading: This is a reported bug that unfortunately can't be fixed easily without breakage on Scala's side. For further information check lampepfl/dotty#13549
The resulting code is still Scala 2 valid code and arguably more correct.
Instead of doing all the changes in an enormous PR with all the changes at once, we can already perform the changes we know for a fact that are going to be needed for Scala 3.
If desired I can provide 1 PR per commit.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] SethTisue commented on pull request #11432: KAFKA-13399 towards scala3
Posted by GitBox <gi...@apache.org>.
SethTisue commented on pull request #11432:
URL: https://github.com/apache/kafka/pull/11432#issuecomment-1044840351
As an aside: once this lands it would be great to get Kafka in the Scala 3 community build. I think VirtusLab hopes to support other build tools besides sbt there eventually, but if that hasn't happened, https://github.com/ennru/kafka/tree/build-with-sbt could probably be brought up to date — that's what we use to [include Kafka](https://github.com/scala/community-build/blob/2.13.x/proj/kafka.conf) in the Scala 2 community build.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] SethTisue edited a comment on pull request #11432: KAFKA-13399 towards scala3
Posted by GitBox <gi...@apache.org>.
SethTisue edited a comment on pull request #11432:
URL: https://github.com/apache/kafka/pull/11432#issuecomment-1044840351
As an aside: once this lands it would be great to get Kafka in the Scala 3 community build. (Among other reasons, because there is a lot of Java code in Kafka, and the Java-consumes-Scala scenario tends to be under-tested in both the 2 and 3 community builds.)
I think VirtusLab hopes to support other build tools besides sbt there eventually, but if that hasn't happened, https://github.com/ennru/kafka/tree/build-with-sbt could probably be brought up to date — that's what we use to [include Kafka](https://github.com/scala/community-build/blob/2.13.x/proj/kafka.conf) in the Scala 2 community build.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jlprat commented on a change in pull request #11432: KAFKA-13399 towards scala3
Posted by GitBox <gi...@apache.org>.
jlprat commented on a change in pull request #11432:
URL: https://github.com/apache/kafka/pull/11432#discussion_r735730354
##########
File path: core/src/main/scala/kafka/raft/KafkaMetadataLog.scala
##########
@@ -42,11 +42,11 @@ final class KafkaMetadataLog private (
// Access to this object needs to be synchronized because it is used by the snapshotting thread to notify the
// polling thread when snapshots are created. This object is also used to store any opened snapshot reader.
snapshots: mutable.TreeMap[OffsetAndEpoch, Option[FileRawSnapshotReader]],
- topicPartition: TopicPartition,
+ topicPartitionArg: TopicPartition,
Review comment:
Avoids a collision between the argument name and the method with the same name.
##########
File path: core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala
##########
@@ -2244,7 +2244,7 @@ class AuthorizerIntegrationTest extends BaseRequestTest {
def testDescribeClusterClusterAuthorizedOperationsWithoutDescribeCluster(): Unit = {
removeAllClientAcls()
- for (version <- ApiKeys.DESCRIBE_CLUSTER.oldestVersion to ApiKeys.DESCRIBE_CLUSTER.latestVersion) {
+ for (version <- ApiKeys.DESCRIBE_CLUSTER.oldestVersion.toInt to ApiKeys.DESCRIBE_CLUSTER.latestVersion.toInt) {
Review comment:
In Scala 2, these `Short`s were converted to `RichInt` to have the `to` method. Now it needs to be done manually.
##########
File path: core/src/test/scala/unit/kafka/coordinator/transaction/TransactionCoordinatorConcurrencyTest.scala
##########
@@ -60,7 +60,7 @@ class TransactionCoordinatorConcurrencyTest extends AbstractCoordinatorConcurren
private val txnRecordsByPartition: Map[Int, mutable.ArrayBuffer[SimpleRecord]] =
(0 until numPartitions).map { i => (i, mutable.ArrayBuffer[SimpleRecord]()) }.toMap
- val producerId = 11
+ val producerId = 11L
Review comment:
This value should be a `Long`, we type it correctly to avoid automatic widening.
##########
File path: core/src/main/scala/kafka/tools/ConsoleConsumer.scala
##########
@@ -27,7 +27,7 @@ import com.typesafe.scalalogging.LazyLogging
import joptsimple._
import kafka.utils.Implicits._
import kafka.utils.{Exit, _}
-import org.apache.kafka.clients.consumer.{Consumer, ConsumerConfig, ConsumerRecord, KafkaConsumer}
+import org.apache.kafka.clients.consumer.{Consumer, ConsumerConfig => ClientConsumerConfig, ConsumerRecord, KafkaConsumer}
Review comment:
This avoids a name clashing between the class `ConsumerConfig` under o.a.k.c.c. with the one defined in the same class.
##########
File path: core/src/main/scala/kafka/admin/TopicCommand.scala
##########
@@ -22,7 +22,7 @@ import java.util.{Collections, Properties}
import joptsimple._
import kafka.common.AdminCommandFailedException
import kafka.log.LogConfig
-import kafka.utils._
+import kafka.utils.{immutable=> _, _}
Review comment:
Changes like this one solve a collision between the `immutable` package in Scala collections and `kafka.utils.immutable`.
##########
File path: core/src/main/scala/kafka/controller/PartitionStateMachine.scala
##########
@@ -477,7 +477,7 @@ class ZkPartitionStateMachine(config: KafkaConfig,
} else {
val (logConfigs, failed) = zkClient.getLogConfigs(
partitionsWithNoLiveInSyncReplicas.iterator.map { case (partition, _) => partition.topic }.toSet,
- config.originals()
+ config.originals
Review comment:
`config.originals` returns a `Map`. In Scala 2 the extra parenthesis were ignored, but in Scala 3 they matter and they are used on the returning `Map`, causing a compilation error. It tries to call `apply` without any parameter.
##########
File path: core/src/main/scala/kafka/utils/json/DecodeJson.scala
##########
@@ -85,13 +85,13 @@ object DecodeJson {
else decodeJson.decodeEither(node).map(Some(_))
}
- implicit def decodeSeq[E, S[+T] <: Seq[E]](implicit decodeJson: DecodeJson[E], factory: Factory[E, S[E]]): DecodeJson[S[E]] = (node: JsonNode) => {
+ implicit def decodeSeq[E, S[E] <: Seq[E]](implicit decodeJson: DecodeJson[E], factory: Factory[E, S[E]]): DecodeJson[S[E]] = (node: JsonNode) => {
Review comment:
This is the only way I could have it work. It seems this definition should have always been like this, but maybe I'm missing a use case.
The previous definition was never met in Scala 3.
##########
File path: core/src/main/scala/kafka/tools/ConsoleProducer.scala
##########
@@ -27,7 +27,7 @@ import kafka.message._
import kafka.utils.Implicits._
import kafka.utils.{CommandDefaultOptions, CommandLineUtils, Exit, ToolsUtils}
import org.apache.kafka.clients.producer.internals.ErrorLoggingCallback
-import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
+import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig => ClientProducerConfig, ProducerRecord}
Review comment:
This avoids a name clashing between the class `ProducerConfig` under o.a.k.c.p. with the one defined in the same class.
##########
File path: core/src/main/scala/kafka/tools/ConsoleConsumer.scala
##########
@@ -61,7 +61,7 @@ object ConsoleConsumer extends Logging {
}
def run(conf: ConsumerConfig): Unit = {
- val timeoutMs = if (conf.timeoutMs >= 0) conf.timeoutMs else Long.MaxValue
+ val timeoutMs: Long = if (conf.timeoutMs >= 0) conf.timeoutMs else Long.MaxValue
Review comment:
`conf.timeoutMs` is an `Int`, so we need to manually type the storing variable with a `Long`.
##########
File path: core/src/test/scala/other/kafka/ReplicationQuotasTestRig.scala
##########
@@ -303,8 +303,8 @@ object ReplicationQuotasTestRig {
s"<p>- BrokerCount: ${config.brokers}" +
s"<p>- PartitionCount: ${config.partitions}" +
f"<p>- Throttle: ${config.throttle.toDouble}%,.0f MB/s" +
- f"<p>- MsgCount: ${config.msgsPerPartition}%,.0f " +
- f"<p>- MsgSize: ${config.msgSize}%,.0f" +
+ f"<p>- MsgCount: ${config.msgsPerPartition.toDouble}%,.0f " +
+ f"<p>- MsgSize: ${config.msgSize.toDouble}%,.0f" +
Review comment:
Those numbers were all converted automatically to `Double`. Now it needs to be done manually.
##########
File path: core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala
##########
@@ -2374,7 +2374,7 @@ class GroupMetadataManagerTest {
minOneMessage = EasyMock.eq(true)))
.andReturn(FetchDataInfo(LogOffsetMetadata(startOffset), mockRecords))
EasyMock.expect(replicaManager.getLog(groupMetadataTopicPartition)).andStubReturn(Some(logMock))
- EasyMock.expect(replicaManager.getLogEndOffset(groupMetadataTopicPartition)).andStubReturn(Some(18))
+ EasyMock.expect(replicaManager.getLogEndOffset(groupMetadataTopicPartition)).andStubReturn(Some(18L))
Review comment:
Method should return `Option[Long]` so we type the `18` correctly.
##########
File path: core/src/main/scala/kafka/server/ZkAdminManager.scala
##########
@@ -874,7 +874,7 @@ class ZkAdminManager(val config: KafkaConfig,
props.setProperty(op.key, value.toString)
case ConfigDef.Type.LONG | ConfigDef.Type.INT =>
val epsilon = 1e-6
- val intValue = if (key.`type` == ConfigDef.Type.LONG)
+ val intValue: Long = if (key.`type` == ConfigDef.Type.LONG)
Review comment:
This was an automatic widening in Scala 2, now needs to be done manually.
##########
File path: core/src/main/scala/kafka/log/LogManager.scala
##########
@@ -353,7 +353,7 @@ class LogManager(logDirs: Seq[File],
s"$logDirAbsolutePath, resetting to the base offset of the first segment", e)
}
- val logsToLoad = Option(dir.listFiles).getOrElse(Array.empty).filter(logDir =>
+ val logsToLoad = Option(dir.listFiles).getOrElse(Array.empty[File]).filter(logDir =>
Review comment:
Scala 3 wasn't able to infer that `Array.empty` parametrized type should be `File`.
##########
File path: core/src/test/scala/unit/kafka/server/FinalizedFeatureChangeListenerTest.scala
##########
@@ -239,10 +239,11 @@ class FinalizedFeatureChangeListenerTest extends ZooKeeperTestHarness {
val exitLatch = new CountDownLatch(1)
Exit.setExitProcedure((_, _) => exitLatch.countDown())
+ val feature1: SupportedVersionRange = brokerFeatures.supportedFeatures.get("feature_1")
Review comment:
This intermediate variable is needed because Scala 3 typer was inferring it to be `BaseVersionRange` which is package protected for `org.apache.kafka.common.feature`.
##########
File path: core/src/test/scala/unit/kafka/server/AbstractCreateTopicsRequestTest.scala
##########
@@ -111,7 +111,7 @@ abstract class AbstractCreateTopicsRequestTest extends BaseRequestTest {
val replication = if (!topic.assignments().isEmpty)
topic.assignments().iterator().next().brokerIds().size()
else
- topic.replicationFactor
+ topic.replicationFactor.toInt
Review comment:
`topic.replicationFactor` is a `Short`, we manually widen it to `Int`.
##########
File path: core/src/main/scala/kafka/admin/ReassignPartitionsCommand.scala
##########
@@ -238,14 +238,14 @@ object ReassignPartitionsCommand extends Logging {
executeAssignment(adminClient,
opts.options.has(opts.additionalOpt),
Utils.readFileAsString(opts.options.valueOf(opts.reassignmentJsonFileOpt)),
- opts.options.valueOf(opts.interBrokerThrottleOpt),
- opts.options.valueOf(opts.replicaAlterLogDirsThrottleOpt),
- opts.options.valueOf(opts.timeoutOpt))
+ opts.options.valueOf(opts.interBrokerThrottleOpt).longValue(),
+ opts.options.valueOf(opts.replicaAlterLogDirsThrottleOpt).longValue(),
+ opts.options.valueOf(opts.timeoutOpt).longValue())
} else if (opts.options.has(opts.cancelOpt)) {
cancelAssignment(adminClient,
Utils.readFileAsString(opts.options.valueOf(opts.reassignmentJsonFileOpt)),
opts.options.has(opts.preserveThrottlesOpt),
- opts.options.valueOf(opts.timeoutOpt))
+ opts.options.valueOf(opts.timeoutOpt).longValue())
Review comment:
Those numbers were all `Int`s converted automatically to `Long`.
##########
File path: core/src/main/scala/kafka/utils/json/DecodeJson.scala
##########
@@ -85,13 +85,13 @@ object DecodeJson {
else decodeJson.decodeEither(node).map(Some(_))
}
- implicit def decodeSeq[E, S[+T] <: Seq[E]](implicit decodeJson: DecodeJson[E], factory: Factory[E, S[E]]): DecodeJson[S[E]] = (node: JsonNode) => {
+ implicit def decodeSeq[E, S[E] <: Seq[E]](implicit decodeJson: DecodeJson[E], factory: Factory[E, S[E]]): DecodeJson[S[E]] = (node: JsonNode) => {
if (node.isArray)
decodeIterator(node.elements.asScala)(decodeJson.decodeEither)
else Left(s"Expected JSON array, received $node")
}
- implicit def decodeMap[V, M[K, +V] <: Map[K, V]](implicit decodeJson: DecodeJson[V], factory: Factory[(String, V), M[String, V]]): DecodeJson[M[String, V]] = (node: JsonNode) => {
+ implicit def decodeMap[V, M[K, V] <: Map[K, V]](implicit decodeJson: DecodeJson[V], factory: Factory[(String, V), M[String, V]]): DecodeJson[M[String, V]] = (node: JsonNode) => {
Review comment:
Fixed this just in case, but there is no test code using this.
##########
File path: core/src/test/scala/unit/kafka/server/IsrExpirationTest.scala
##########
@@ -59,7 +59,7 @@ class IsrExpirationTest {
@BeforeEach
def setUp(): Unit = {
val logManager: LogManager = EasyMock.createMock(classOf[LogManager])
- EasyMock.expect(logManager.liveLogDirs).andReturn(Array.empty[File]).anyTimes()
+ EasyMock.expect(logManager.liveLogDirs).andReturn(Seq.empty[File]).anyTimes()
Review comment:
Method is expected to return a `Seq` instead of `Array`. Previous code wasn't working in Scala 3.
##########
File path: core/src/main/scala/kafka/raft/KafkaNetworkChannel.scala
##########
@@ -150,7 +150,7 @@ class KafkaNetworkChannel(
request.createdTimeMs,
destination = node,
request = buildRequest(request.data),
- handler = onComplete
+ handler = onComplete(_)
Review comment:
A more idiomatic way to pass the function reference. Previous wasn't compiling in Scala 3.
##########
File path: core/src/main/scala/kafka/log/LogConfig.scala
##########
@@ -172,7 +172,7 @@ case class LogConfig(props: java.util.Map[_, _], overriddenConfigs: Set[String]
}
def overriddenConfigsAsLoggableString: String = {
- val overriddenTopicProps = props.asScala.collect {
+ val overriddenTopicProps: Map[String, Object] = props.asScala.collect {
Review comment:
For some reason Scala 3 was picking `Map[AnyRef, AnyRef]` instead.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jlprat commented on pull request #11432: KAFKA-13399 towards scala3
Posted by GitBox <gi...@apache.org>.
jlprat commented on pull request #11432:
URL: https://github.com/apache/kafka/pull/11432#issuecomment-951077960
cc. @ijuma if you have some time to review this I would be thankful!
These are the changes I mentioned in the mailing list in regards to the Scala 3 migration. These are only the ones that are for sure needed. Other changes that were present in https://github.com/apache/kafka/pull/11350 but not here might not be needed once we migrate to Scala 3 as they are either already solved and will be in future versions of Scala, or being worked on.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jlprat commented on pull request #11432: KAFKA-13399 towards scala3
Posted by GitBox <gi...@apache.org>.
jlprat commented on pull request #11432:
URL: https://github.com/apache/kafka/pull/11432#issuecomment-1044848322
Hi @SethTisue thanks! I definitely will take a look at it!
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jlprat commented on pull request #11432: KAFKA-13399 towards scala3
Posted by GitBox <gi...@apache.org>.
jlprat commented on pull request #11432:
URL: https://github.com/apache/kafka/pull/11432#issuecomment-977013245
Hi @cmccabe and @ijuma, I would highly appreciate your feedback on this PR.
Thanks in advance!
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jlprat commented on pull request #11432: KAFKA-13399 towards scala3
Posted by GitBox <gi...@apache.org>.
jlprat commented on pull request #11432:
URL: https://github.com/apache/kafka/pull/11432#issuecomment-951174972
Test failure was https://issues.apache.org/jira/browse/KAFKA-4184
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jlprat commented on pull request #11432: KAFKA-13399 towards scala3
Posted by GitBox <gi...@apache.org>.
jlprat commented on pull request #11432:
URL: https://github.com/apache/kafka/pull/11432#issuecomment-974062052
Error was a flaky test: https://issues.apache.org/jira/browse/KAFKA-8785
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org