KafkaRequestActionBuilder.scala

class KafkaRequestActionBuilder[K, V <: Message](
  topic: String,  //this parameters will be provided when building KafkaPublisher
  keyExtractor: V => Option[K]) extends RequestActionBuilder[V] {
  
  override type PK = KafkaProtocol.KafkaProtocolKey.type
  override def key = KafkaProtocol.KafkaProtocolKey

  override def build(ctx: ScenarioContext, components: KafkaComponents): Producer[V] = {
    val producer = new KafkaProducer[K, V](components.protocol.properties.asJava)
    new KafkaPublisher[K, V](producer, topic)(keyExtractor)
  }
}