Skip to content
This repository has been archived by the owner on Mar 22, 2022. It is now read-only.

Feature/kafka serializers #24

Open
wants to merge 3 commits into
base: master
Choose a base branch
from
Open

Feature/kafka serializers #24

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
9dfc011
Extended the kafka protocol with the ability to explicitly specify se…
Apr 23, 2019
ad43711
Using the correct KafkaProducer constructor.
Apr 23, 2019
3bdde9c
Casting the protocol serializers to the correct type.
Apr 23, 2019
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
10 changes: 7 additions & 3 deletions src/main/scala/com/github/mnogu/gatling/kafka/action/KafkaRequestActionBuilder.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,18 +6,22 @@ import io.gatling.core.action.Action
import io.gatling.core.action.builder.ActionBuilder
import io.gatling.core.structure.ScenarioContext
import org.apache.kafka.clients.producer.KafkaProducer
import org.apache.kafka.common.serialization.Serializer

import scala.collection.JavaConverters._


class KafkaRequestActionBuilder[K,V](kafkaAttributes: KafkaAttributes[K,V]) extends ActionBuilder {
class KafkaRequestActionBuilder[K, V](kafkaAttributes: KafkaAttributes[K, V]) extends ActionBuilder {

override def build( ctx: ScenarioContext, next: Action ): Action = {
import ctx.{protocolComponentsRegistry, coreComponents, throttled}

val kafkaComponents: KafkaComponents = protocolComponentsRegistry.components(KafkaProtocol.KafkaProtocolKey)

val producer = new KafkaProducer[K,V]( kafkaComponents.kafkaProtocol.properties.asJava )
val producer = new KafkaProducer[K, V](
kafkaComponents.kafkaProtocol.properties.asJava,
kafkaComponents.kafkaProtocol.keySerializerOpt.asInstanceOf[Option[Serializer[K]]].orNull,
kafkaComponents.kafkaProtocol.valueSerializerOpt.asInstanceOf[Option[Serializer[V]]].orNull
)

coreComponents.actorSystem.registerOnTermination(producer.close())

Expand Down
11 changes: 9 additions & 2 deletions src/main/scala/com/github/mnogu/gatling/kafka/protocol/KafkaProtocol.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,12 +3,15 @@ package com.github.mnogu.gatling.kafka.protocol
import io.gatling.core.CoreComponents
import io.gatling.core.config.GatlingConfiguration
import io.gatling.core.protocol.{Protocol, ProtocolKey}
import org.apache.kafka.common.serialization.Serializer

object KafkaProtocol {

def apply(configuration: GatlingConfiguration): KafkaProtocol = KafkaProtocol (
topic = "",
properties = Map()
properties = Map(),
None,
None
)

val KafkaProtocolKey = new ProtocolKey[KafkaProtocol, KafkaComponents] {
Expand All @@ -35,8 +38,12 @@ object KafkaProtocol {

case class KafkaProtocol(
topic: String,
properties: Map[String, Object]) extends Protocol {
properties: Map[String, Object],
keySerializerOpt: Option[Serializer[_]],
valueSerializerOpt: Option[Serializer[_]]) extends Protocol {

def topic(topic: String): KafkaProtocol = copy(topic = topic)
def properties(properties: Map[String, Object]): KafkaProtocol = copy(properties = properties)
def keySerializer(keySerializer: Serializer[_]): KafkaProtocol = copy(keySerializerOpt = Some(keySerializer))
def valueSerializer(valueSerializer: Serializer[_]): KafkaProtocol = copy(valueSerializerOpt = Some(valueSerializer))
}