[DPE-7148] feat: KIP-714 PoC

src/charm.py

@@ -32,6 +32,7 @@
     CHARM_KEY,
     JMX_CC_PORT,
     JMX_EXPORTER_PORT,
+    KIP714_LIBPATH,
     LOGS_RULES_DIR,
     METRICS_RULES_DIR,
     OS_REQUIREMENTS,
@@ -118,6 +119,7 @@ def _on_install(self, _) -> None:
             return
 
         self._set_os_config()
+        self.workload.exec(f"cp src/{KIP714_LIBPATH} {self.workload.paths.data_path}")
 
     def _set_os_config(self) -> None:
         """Sets sysctl config."""

src/literals.py

@@ -83,7 +83,8 @@ class TLSScope(str, Enum):
 JMX_CC_PORT = 9102
 METRICS_RULES_DIR = "./src/alert_rules/prometheus"
 LOGS_RULES_DIR = "./src/alert_rules/loki"
-
+KIP714_LIBPATH = "custom-kafka-metric-exporter-1.0.jar"
+KIP714_CLASSNAME = "org.imandata.kafka.CustomMetricsReporter"
 
 @dataclass
 class Ports:

src/managers/config.py

@@ -32,6 +32,8 @@
     JMX_EXPORTER_PORT,
     JVM_MEM_MAX_GB,
     JVM_MEM_MIN_GB,
+    KIP714_CLASSNAME,
+    KIP714_LIBPATH,
     KRAFT_NODE_ID_OFFSET,
     PATHS,
     PROFILE_TESTING,
@@ -792,6 +794,7 @@ def server_properties(self) -> list[str]:
                 f"listeners={','.join(listeners_repr)}",
                 f"advertised.listeners={','.join(advertised_listeners)}",
                 f"inter.broker.listener.name={self.internal_listener.name}",
+                f"metric.reporters={KIP714_CLASSNAME}"
             ]
             + self.scram_properties
             + self.oauth_properties
@@ -848,6 +851,8 @@ def set_environment(self) -> None:
             self.jvm_performance_opts,
             self.heap_opts,
             self.log_level,
+            # TODO: ideally, this should be bundled in the snap
+            f'CLASSPATH={PATHS["kafka"]["DATA"]}/{KIP714_LIBPATH}',
         ] + self.auxiliary_paths
 
         raw_current_env = self.workload.read("/etc/environment")