kafka_filter.scala

val filteredEvents: KStream[Integer, (HeartRateEvent, BloodPressureEvent)] =
  combinedEvents
    .filter((userId: Integer, tuple: (Option[HeartRateEvent], Option[BloodPressureEvent])) => {
      (userId, tuple) match {
        case (_, (Some(hrEvent: HeartRateEvent), Some(bpEvent: BloodPressureEvent))) =>
          hrEvent.heartRate > 100 && bpEvent.systolic < 100
        case _ => false
      }
    }).mapValues {
      case (Some(hrEvent: HeartRateEvent), Some(bpEvent: BloodPressureEvent)) => (hrEvent, bpEvent)
    }