We currently only expose the combined watermark in the kafka source in the stats(). We want to expose per-partition watermark information in verbose stats as well.

In the KafkaConsumerOperator, there is a watermark for each input partition.

We can expose the partition watermark as a "watermark" field within the KafkaConsumerPartitionState struct:

https://github.com/10gen/mongo/blob/master/src/mongo/db/modules/enterprise/src/streams/commands/stream_ops.idl#L134

We can test this new functionality in the below places:

• kafka_consumer_operator_test (run with ninja +kafka_consumer_operator_test -j400). This tests using mock Kafka classes.
• kafka.js test (run with s-mongo-build && test-streams-kafka). This tests against a real kafka broker running in a local docker container.