[kafkajs]: Instrument Consumer

[bizob2828]

Description

We should add instrumentation to support eachMessage. See docs

Consumer

Each time a message is consumed by the consumer, a transaction MUST be
started/stopped. If possible, a transaction SHOULD be started and stopped around a
single message. If not possible, a transaction MUST be started and stopped around
multiple messages.

Transaction duration MUST include time of the following:

• The customer's handling of the message(s).
  Transaction duration MAY include time of the following:
• Open a connection as needed
• Fetch the data
• Close the connection as needed

Transaction Name

The transaction name MUST be in the following format:
Message/Kafka/Topic/Consume/Named/{topic_name}

Metric Names

The following metrics MUST be captured for each transaction and the metric names
MUST be in the following format:

Attributes	Description
Message/Kafka/Topic/Named/{topic_name}/Received/Bytes	The number of Bytes received in the message for each topic.
Message/Kafka/Topic/Named/{topic_name}/Received/Messages	The number of messages received for each topic.

Agent Attributes

The following attributes SHOULD be included in each transaction:

Attributes	Description
kafka.consume.byteCount	The number of Bytes received in the request.
kafka.consume.client_id	The client id of the consumer receiving the data.

Deserialization Metrics

Because serializing and deserializing is such a common thing to do with Kafka, many
implementations include this as an option directly on the producer/consumer or as a
specialized Serializingproducer Deserializingconsumer. It very useful to capture the
time it takes to deserialize the consumer's message as there may be significant slow
downs due to this type of operation. A message consists of both a key and a value which
are deserialized independently. If deserialization monitoring is possible, the following
metrics MUST be captured:

• MessageBroker/Kafka/Topic/Named/{topic_name}/Serialization/Value
• MessageBroker/Kafka/Topic/Named/{topic_name}/Serialization/Key

Additional context

• The deserialization metrics may not be possible with kafkajs.
• We should be able to use the message-shim recordConsume method with the appropriate spec

[workato-integration]

https://new-relic.atlassian.net/browse/NR-274407