implemented querying for Kafka via BDQL