我使用playframework（scala）作为微服务，并使用kafka作为事件总线。我有一个Map到如下事件类的事件使用者：

case class MovieEvent[T] (
                          mediaId: String,
                          config: T
                        )

object MovieEvent {
  implicit def movieEventFormat[T: Format]: Format[MovieEvent[T]] =
    ((__ \ "mediaId").format[String] ~
      (__ \ "config").format[T]
      )(MovieEvent.apply _, unlift(MovieEvent.unapply))
}

object MovieProvider extends SerializableEnumeration {
  implicit val providerReads: Reads[MovieProvider.Value] = SerializableEnumeration.jsonReader(MovieProvider)
  implicit val providerWrites: Writes[MovieProvider.Value] = SerializableEnumeration.jsonWrites
  val  Dreamworks, Disney, Paramount = Value
}

消费者看起来像：

class MovieEventConsumer @Inject()(movieService: MovieService
                                    ) extends ConsumerRecordProcessor with LazyLogging {
  override def process(record: IncomingRecord): Unit = {

    val movieEventJson = Json.parse(record.valueString).validate[MovieEvent[DreamworksConfiguration]]
    movieEventJson match {
      case event: JsSuccess[MovieEvent[DreamworksJobOptions]] => processMovieEvent(event.get)
      case er: JsError =>
        logger.error("Unrecognized MovieEvent, attempting to parse as MovieUploadEvent: " + JsError.toJson(er).toString())
        try {
          val data = (Json.parse(record.valueString) \ "upload").as[MovieUploadEvent]
          processUploadEvent(data)
        } catch {
          case er: Exception => logger.error("Unrecognized kafka event", er)
        }
    }
  }

  def processMovieEvent[T](event: MovieEvent[T]): Unit = {
    logger.debug(s"Received movie event: ${event}")
    movieService.createMovieJob(event)
  }

  def processUploadEvent(event: MovieUploadEvent): Unit = {
    logger.debug(s"Received upload event: ${event}")
    movieService.addToCollection(event)
  }

}

现在，我只能验证三种不同的movieevent配置（梦工厂、迪斯尼和派拉蒙）中的一种。我可以交换我通过代码验证的那一个，但这不是重点。但是，我想验证这三种方法中的任何一种，而不必增加消费者。我试过玩弄一些不同的想法，但都没有。我对Kafka和Kafka还很陌生，不知道有没有一个好的方法可以做到这一点。
提前谢谢！