function handler({ consumer }) {
  /**
   * Message handler that extracts the topic and headers from message being consumed.
   *
   * This also sets some metrics for byte length of message, and number of messages.
   * Lastly, adds tx attributes for byteCount and clientId
   *
   * @param {MessageShim} shim instance of shim
   * @param {Array} args arguments passed to the `eachMessage` function of the `consumer.run`
   * @param {Transaction} tx active transaction
   * @returns {MessageSpec} spec for message handling
   */
  return function messageHandler(shim, args, tx) {
    recordMethodMetric({ agent: shim.agent, name: 'eachMessage' })

    const [data] = args
    const { topic } = data

    recordLinkingMetrics({
      agent: shim.agent,
      brokers: consumer[kafkaCtx].brokers,
      topic,
      producer: false
    })

    if (tx) {
      const byteLength = data?.message.value?.byteLength
      const metricPrefix = `Message/Kafka/Topic/Named/${topic}/Received`
      // This will always be 1
      tx.metrics.getOrCreateMetric(`${metricPrefix}/Messages`).recordValue(1)
      if (byteLength) {
        tx.metrics.measureBytes(`${metricPrefix}/Bytes`, byteLength)
        tx.trace.attributes.addAttribute(
          DESTINATIONS.TRANS_SCOPE,
          'kafka.consume.byteCount',
          byteLength
        )
      }
      if (consumer?.[kafkaCtx]) {
        tx.trace.attributes.addAttribute(
          DESTINATIONS.TRANS_EVENT,
          'kafka.consume.client_id',
          consumer[kafkaCtx].clientId
        )
      }
    }

    return new MessageSpec({
      destinationType: 'Topic/Consume',
      destinationName: data?.topic,
      headers: data?.message?.headers
    })
  }
}
Wrapped of message handler that passes in consumer and recordMethodMetric.
We do not want to bind this as this gets called with the appropriate binding
in message-shim

@param {object} params to function
@param {object} params.consumer consumer being instrumented
@returns {function} message handler for setting metrics and spec for the consumer transaction