...

1. Topic name

2. Client principal

3. Client IP address

4. Number of bytes produced or consumed

5. Records count (if the batch version is 2 or greater, otherwise a deep decompression is required to get the count)

Proposed Changes

Add interfaces and implementations for classes that are described above and

...

a configuration property in the KafkaConfig class to allow users to specify implementations of the observer interface. The configuration property lets user define a list of observer implementations which are going to be invoked.

/** ********* Broker-side Observer Configuration ****************/
val ObserverClassProp = "observer.class.names"

...