Today, an Atlas Stream Processing user can specify $emit.config.headers = expression. The expression must resolve to a document or array of key value pairs. Today, each value in the document or k/v array must be a binData.

We'd like to implement the behavior below as discussed in the product description:

Each value in the $emit.config.headers document or array must be a string, document, binData, int, long, double, or null/missing. We automatically serialize each of these types into utf8 strings, utf8 json, binData, 4 byte big endian integers, 8 byte big endian integers, 8 byte IEEE 754 big endian double, and empty byte array, respectively.

The relevant code changes will be in this class:

https://github.com/10gen/mongo/blob/master/src/mongo/db/modules/enterprise/src/streams/exec/kafka_emit_operator.cpp

We can add tests in kafka.js:

https://github.com/10gen/mongo/blob/master/src/mongo/db/modules/enterprise/jstests/streams_kafka/kafka.js