Skip to content

upgrade kafka client library to 2.8.2 #5400

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

Merged
merged 5 commits into from
May 3, 2023
Merged

upgrade kafka client library to 2.8.2 #5400

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
b8c4cba
upgrade kafka client library
Apr 27, 2023
b9d1c12
attempt build upgrading embedded kafka
Apr 27, 2023
5dc416a
attempt to fix standalone server startup test
May 2, 2023
cc2bcb0
bump to kafka client to latest patch
May 2, 2023
2f6bdba
revert kafka test timeout config change
May 3, 2023
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
2 changes: 1 addition & 1 deletion common/scala/build.gradle
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,7 +61,7 @@ dependencies {
api "commons-codec:commons-codec:1.9"
api "commons-io:commons-io:2.11.0"
api "commons-collections:commons-collections:3.2.2"
api "org.apache.kafka:kafka-clients:2.4.0"
api "org.apache.kafka:kafka-clients:2.8.2"
api "org.apache.httpcomponents:httpclient:4.5.5"
api "com.fasterxml.uuid:java-uuid-generator:3.1.3"
api "com.github.ben-manes.caffeine:caffeine:2.6.2"
Expand Down
4 changes: 2 additions & 2 deletions core/monitoring/user-events/build.gradle
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,9 +45,9 @@ dependencies {

testImplementation "junit:junit:4.11"
testImplementation "org.scalatest:scalatest_${gradle.scala.depVersion}:3.0.8"
testImplementation "io.github.embeddedkafka:embedded-kafka_${gradle.scala.depVersion}:2.4.0"
testImplementation "io.github.embeddedkafka:embedded-kafka_${gradle.scala.depVersion}:2.8.1"
Copy link
Contributor Author

Choose a reason for hiding this comment

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

embedded kafka for the tests doesn't have a release for 2.8.2, but works against any 2.8.x version of the kafka client.

constraints {
testImplementation("io.github.embeddedkafka:embedded-kafka_${gradle.scala.depVersion}:2.4.0")
testImplementation("io.github.embeddedkafka:embedded-kafka_${gradle.scala.depVersion}:2.8.1")
testImplementation('org.apache.avro:avro:1.11.1') {
because 'embeddedkafka dependency cannot be upgraded currently and avro in embedded kafka 2.4.0 has vulns'
}
Expand Down
20 changes: 17 additions & 3 deletions ...er-events/src/test/scala/org/apache/openwhisk/core/monitoring/metrics/KafkaSpecBase.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,8 +17,8 @@

package org.apache.openwhisk.core.monitoring.metrics

import akka.kafka.testkit.scaladsl.{EmbeddedKafkaLike, ScalatestKafkaSpec}
import net.manub.embeddedkafka.EmbeddedKafka
import akka.kafka.testkit.scaladsl.ScalatestKafkaSpec
import io.github.embeddedkafka.{EmbeddedKafka, EmbeddedKafkaConfig}
import org.scalatest._
import org.scalatest.concurrent.{Eventually, IntegrationPatience, ScalaFutures}

Expand All @@ -30,11 +30,25 @@ abstract class KafkaSpecBase
with ScalaFutures
with FlatSpecLike
with EmbeddedKafka
with EmbeddedKafkaLike
with IntegrationPatience
with Eventually
with EventsTestHelper { this: Suite =>
implicit val timeoutConfig: PatienceConfig = PatienceConfig(1.minute)
override val sleepAfterProduce: FiniteDuration = 10.seconds
override protected val topicCreationTimeout = 60.seconds
override protected val producerPublishTimeout: FiniteDuration = 60.seconds

lazy implicit val embeddedKafkaConfig: EmbeddedKafkaConfig = EmbeddedKafkaConfig(kafkaPort, zooKeeperPort)

override def bootstrapServers = s"localhost:${embeddedKafkaConfig.kafkaPort}"

override def setUp(): Unit = {
EmbeddedKafka.start()(embeddedKafkaConfig)
super.setUp()
}

override def cleanUp(): Unit = {
super.cleanUp()
EmbeddedKafka.stop()
}
}
4 changes: 2 additions & 2 deletions core/standalone/build.gradle
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -169,9 +169,9 @@ dependencies {
implementation project(':tools:admin')
implementation "org.rogach:scallop_${gradle.scala.depVersion}:3.3.2"

implementation "io.github.embeddedkafka:embedded-kafka_${gradle.scala.depVersion}:2.4.0"
implementation "io.github.embeddedkafka:embedded-kafka_${gradle.scala.depVersion}:2.8.1"
constraints {
implementation("io.github.embeddedkafka:embedded-kafka_${gradle.scala.depVersion}:2.4.0")
implementation("io.github.embeddedkafka:embedded-kafka_${gradle.scala.depVersion}:2.8.1")
implementation('org.apache.avro:avro:1.11.1') {
because 'embeddedkafka dependency cannot be upgraded currently and avro in embedded kafka 2.4.0 has vulns'
}
Expand Down
10 changes: 6 additions & 4 deletions core/standalone/src/main/scala/org/apache/openwhisk/standalone/KafkaLauncher.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,10 +18,9 @@
package org.apache.openwhisk.standalone

import java.io.File

import akka.actor.ActorSystem
import kafka.server.KafkaConfig
import net.manub.embeddedkafka.{EmbeddedKafka, EmbeddedKafkaConfig}
import io.github.embeddedkafka.{EmbeddedKafka, EmbeddedKafkaConfig}
import org.apache.commons.io.FileUtils
import org.apache.openwhisk.common.{Logging, TransactionId}
import org.apache.openwhisk.core.WhiskConfig
Expand All @@ -30,6 +29,7 @@ import org.apache.openwhisk.core.entity.ControllerInstanceId
import org.apache.openwhisk.core.loadBalancer.{LeanBalancer, LoadBalancer, LoadBalancerProvider}
import org.apache.openwhisk.standalone.StandaloneDockerSupport.{checkOrAllocatePort, containerName, createRunCmd}

import java.nio.file.FileSystems
import scala.concurrent.{ExecutionContext, Future}
import scala.reflect.io.Directory
import scala.util.Try
Expand Down Expand Up @@ -66,8 +66,10 @@ class KafkaLauncher(
EmbeddedKafkaConfig(kafkaPort = kafkaPort, zooKeeperPort = zkPort, customBrokerProperties = brokerProps)

val t = Try {
EmbeddedKafka.startZooKeeper(createDir("zookeeper"))
EmbeddedKafka.startKafka(createDir("kafka"))
createDir("zookeeper")
createDir("kafka")
EmbeddedKafka.startZooKeeper(FileSystems.getDefault.getPath(workDir.getPath,"zookeeper"))
EmbeddedKafka.startKafka(FileSystems.getDefault.getPath(workDir.getPath,"kafka"))
}

Future
Expand Down
2 changes: 1 addition & 1 deletion settings.gradle
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,7 +78,7 @@ gradle.ext.scalafmt = [
]

gradle.ext.akka = [version : '2.6.12']
gradle.ext.akka_kafka = [version : '2.0.2']
gradle.ext.akka_kafka = [version : '2.0.5']
gradle.ext.akka_http = [version : '10.2.4']
gradle.ext.akka_management = [version : '1.0.5']

Expand Down
5 changes: 2 additions & 3 deletions tests/build.gradle
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -239,9 +239,9 @@ dependencies {
because 'swagger-request-validator-core cannot be upgraded to 2.x where vuln is remediated'
}
}
implementation "io.github.embeddedkafka:embedded-kafka_${gradle.scala.depVersion}:2.4.0"
implementation "io.github.embeddedkafka:embedded-kafka_${gradle.scala.depVersion}:2.8.1"
constraints {
implementation("io.github.embeddedkafka:embedded-kafka_${gradle.scala.depVersion}:2.4.0")
implementation("io.github.embeddedkafka:embedded-kafka_${gradle.scala.depVersion}:2.8.1")
implementation('org.apache.avro:avro:1.11.1') {
because 'embeddedkafka dependency cannot be upgraded currently and avro in embedded kafka 2.4.0 has vulns'
}
Expand All @@ -255,7 +255,6 @@ dependencies {
implementation "com.microsoft.azure:azure-cosmos:3.7.6"
implementation 'org.testcontainers:elasticsearch:1.17.6'
implementation 'org.testcontainers:mongodb:1.17.1'

implementation project(':common:scala')
implementation project(':core:controller')
implementation project(':core:scheduler')
Expand Down
30 changes: 16 additions & 14 deletions .../test/scala/org/apache/openwhisk/core/database/cosmosdb/cache/CacheInvalidatorTests.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,26 +17,17 @@

package org.apache.openwhisk.core.database.cosmosdb.cache
import java.net.UnknownHostException

import akka.Done
import akka.actor.CoordinatedShutdown
import akka.kafka.testkit.scaladsl.{EmbeddedKafkaLike, ScalatestKafkaSpec}
import akka.kafka.testkit.scaladsl.ScalatestKafkaSpec
import com.typesafe.config.ConfigFactory
import net.manub.embeddedkafka.{EmbeddedKafka, EmbeddedKafkaConfig}
import io.github.embeddedkafka.{EmbeddedKafka, EmbeddedKafkaConfig}
import org.apache.kafka.common.KafkaException
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.openwhisk.common.{AkkaLogging, TransactionId}
import org.apache.openwhisk.core.database.{CacheInvalidationMessage, RemoteCacheInvalidation}
import org.apache.openwhisk.core.database.cosmosdb.{CosmosDBArtifactStoreProvider, CosmosDBTestSupport}
import org.apache.openwhisk.core.entity.{
DocumentReader,
EntityName,
EntityPath,
WhiskDocumentReader,
WhiskEntity,
WhiskEntityJsonFormat,
WhiskPackage
}
import org.apache.openwhisk.core.entity.{DocumentReader, EntityName, EntityPath, WhiskDocumentReader, WhiskEntity, WhiskEntityJsonFormat, WhiskPackage}
import org.junit.runner.RunWith
import org.scalatest.concurrent.ScalaFutures
import org.scalatest.junit.JUnitRunner
Expand All @@ -48,7 +39,6 @@ import scala.util.Random
@RunWith(classOf[JUnitRunner])
class CacheInvalidatorTests
extends ScalatestKafkaSpec(6061)
with EmbeddedKafkaLike
with EmbeddedKafka
with CosmosDBTestSupport
with Matchers
Expand All @@ -58,7 +48,19 @@ class CacheInvalidatorTests
private implicit val logging = new AkkaLogging(system.log)
implicit override val patienceConfig: PatienceConfig = PatienceConfig(timeout = 300.seconds)

override def createKafkaConfig: EmbeddedKafkaConfig = EmbeddedKafkaConfig(kafkaPort, zooKeeperPort)
def createKafkaConfig: EmbeddedKafkaConfig = EmbeddedKafkaConfig(kafkaPort, zooKeeperPort)

override def bootstrapServers = s"localhost:$kafkaPort"

override def setUp(): Unit = {
EmbeddedKafka.start()(createKafkaConfig)
super.setUp()
}

override def cleanUp(): Unit = {
super.cleanUp()
EmbeddedKafka.stop()
}

behavior of "CosmosDB CacheInvalidation"

Expand Down