event_processor.scala

def buildEventStream = {
    import AppSerdes.movieBEventSerde.eventConsumed
    builder.stream[Int, BusinessEvent]("events")
  }

  private val eventStreams: KStreamS[Int, BusinessEvent] = buildEventStream

  def filterEventsByType(eventType: String): KStreamS[Int, BusinessEvent] = {
    eventStreams.filter((_: Int, event: BusinessEvent) => event.eventType.equalsIgnoreCase(eventType))
  }

  filterEventsByType(EventTypes.`MOVIECREATEEVENT`).foreach((id, event) => {

    val value = new MovieCreatedEventDeserializer(event).get
    if (value.isDefined) {
      val doc = value.get
      movies.insertOne(document = doc).toFuture().onComplete(_ => println(s"Inserted ${doc}"))
    }

  })

  filterEventsByType(EventTypes.`MOVIEUPDATEEVENT`).foreach((id, event) => {
    val movie: Option[Document] = new MovieUpdateEventDeserializer(event).get
    val movieDocument           = movies.find(Filters.eq("movie_id", movie.get.getInteger("movie_id")))
    movieDocument.toFuture().map(_.head).onSuccess {
      case data => {
        val document = movie.get.toBsonDocument
        println(s"Relpacing Movie Information ${movie.get.get("movie_id").get.asString()}")
        document.put("sales", data.get("sales").get)
        movies
          .replaceOne(Filters.eq("_id", data.getObjectId("_id")), document)
          .toFuture()
          .onSuccess {
            case data => println("Movie Information Updated")
          }
      }
    }

  })