Version 3.5

Author: Amerousful

README.md

@@ -11,7 +11,7 @@ Add to your `pom.xml`
 <dependency>
     <groupId>io.github.amerousful</groupId>
     <artifactId>gatling-kafka</artifactId>
-    <version>3.4</version>
+    <version>3.5</version>
 </dependency>
 ```
 
@@ -20,7 +20,7 @@ Add to your `pom.xml`
 Add to your `build.sbt`
 
 ```scala
-libraryDependencies += "io.github.amerousful" % "gatling-kafka" % "3.4"
+libraryDependencies += "io.github.amerousful" % "gatling-kafka" % "3.5"
 ```
 
 Import:

build.sbt

@@ -1,6 +1,6 @@
 name := "gatling-kafka"
 
-version := "3.4"
+version := "3.5"
 
 scalaVersion := "2.13.12"
 

changelog.txt

@@ -1,3 +1,8 @@
+Changelog 3.5 (10-11-2024)
+----------------------------
+* Added support for SSL using truststore
+* Fix single initialization of properties for Kafka consumer
+
 Changelog 3.4 (02-08-2024)
 ----------------------------
 * Fix. Remove `seekToEnd` after rebalancing

src/main/java/io/github/amerousful/kafka/javaapi/KafkaProtocolBuilder.java

@@ -90,6 +90,11 @@ public KafkaProtocolBuilder ssl(@NonNull String protocol, @NonNull String keysto
         return new KafkaProtocolBuilder(wrapped.ssl(protocol, keystoreLocation, keystorePassword, keyPassword));
     }
 
+    @NonNull
+    public KafkaProtocolBuilder ssl(@NonNull String protocol, @NonNull String keystoreLocation, @NonNull String keystorePassword, @NonNull String keyPassword, @NonNull String trustStoreLocation, @NonNull String trustStorePassword) {
+        return new KafkaProtocolBuilder(wrapped.ssl(protocol, keystoreLocation, keystorePassword, keyPassword, trustStoreLocation, trustStorePassword));
+    }
+
     @NonNull
     public KafkaProtocolBuilder replyTimeout(long timeout) {
         return replyTimeout(Duration.ofSeconds(timeout));

src/main/scala/io/github/amerousful/kafka/client/KafkaTrackerPoll.scala

@@ -50,7 +50,7 @@ class KafkaTrackerPoll(
     case TrackerAndController(_, consumerControl) => consumerControl.shutdown()
   }
 
-  private val consumerPropertiesTrackerPool: Map[String, AnyRef] = (Map(
+  private def consumerPropertiesTrackerPool: Map[String, AnyRef] = (Map(
     "enable.auto.commit" -> "true",
     "auto.offset.reset" -> "latest"
   ) ++ consumerProperties).updatedWith("group.id")({

src/main/scala/io/github/amerousful/kafka/protocol/KafkaProtocolBuilder.scala

@@ -2,9 +2,9 @@ package io.github.amerousful.kafka.protocol
 
 import com.softwaremill.quicklens.ModifyPimp
 import io.github.amerousful.kafka.protocol.AutoOffsetReset.AutoOffsetReset
-import org.apache.kafka.clients.producer.ProducerConfig._
-import org.apache.kafka.clients.consumer.ConsumerConfig._
 import io.github.amerousful.kafka.protocol.SaslMechanism.SaslMechanism
+import org.apache.kafka.clients.consumer.ConsumerConfig._
+import org.apache.kafka.clients.producer.ProducerConfig._
 
 import scala.concurrent.duration.FiniteDuration
 
@@ -64,6 +64,19 @@ final case class KafkaProtocolBuilder(kafkaProtocol: KafkaProtocol) {
       .addConsumerAndProducerProperty("ssl.key.password", keyPassword)
   }
 
+  def ssl(
+           protocol: String,
+           keystoreLocation: String,
+           keystorePassword: String,
+           keyPassword: String,
+           trustStoreLocation: String,
+           trustStorePassword: String
+         ): KafkaProtocolBuilder = {
+    ssl(protocol, keystoreLocation, keystorePassword, keyPassword)
+      .addConsumerAndProducerProperty("ssl.truststore.location", trustStoreLocation)
+      .addConsumerAndProducerProperty("ssl.truststore.password", trustStorePassword)
+  }
+
   def addProducerProperty(key: String, value: String): KafkaProtocolBuilder = {
     this.modify(_.kafkaProtocol.producerProperties)(_ + (key -> value))
   }
@@ -89,7 +102,7 @@ final case class KafkaProtocolBuilder(kafkaProtocol: KafkaProtocol) {
 
   def schemaUrl(url: String): KafkaProtocolBuilder = addConsumerAndProducerProperty("schema.registry.url", url)
 
-  def disableAutoCommit() =  addConsumerProperty(ENABLE_AUTO_COMMIT_CONFIG, "false")
+  def disableAutoCommit() = addConsumerProperty(ENABLE_AUTO_COMMIT_CONFIG, "false")
 
   def autoOffsetResetPolicy(value: AutoOffsetReset) = addConsumerProperty(AUTO_OFFSET_RESET_CONFIG, value)