Added kafka client id

Author: markretallack

src/karapace/core/kafka_utils.py

@@ -54,6 +54,7 @@ def kafka_consumer_from_config(config: Config, topic: str) -> Iterator[KafkaCons
 def kafka_producer_from_config(config: Config) -> Iterator[KafkaProducer]:
     producer = KafkaProducer(
         bootstrap_servers=config.bootstrap_uri,
+        client_id=config.client_id,
         security_protocol=config.security_protocol,
         ssl_cafile=config.ssl_cafile,
         ssl_certfile=config.ssl_certfile,

src/karapace/core/messaging.py

@@ -39,6 +39,7 @@ def initialize_karapace_producer(
             try:
                 self._producer = KafkaProducer(
                     bootstrap_servers=self._config.bootstrap_uri,
+                    client_id=self._config.client_id,
                     verify_connection=False,
                     security_protocol=self._config.security_protocol,
                     ssl_cafile=self._config.ssl_cafile,

src/karapace/kafka_rest_apis/__init__.py

@@ -525,6 +525,7 @@ async def _maybe_create_async_producer(self) -> AsyncKafkaProducer:
 
                 producer = AsyncKafkaProducer(
                     acks=acks,
+                    client_id=self.config.client_id,
                     bootstrap_servers=self.config.bootstrap_uri,
                     compression_type=self.config.producer_compression_type,
                     connections_max_idle_ms=self.config.connections_max_idle_ms,
@@ -752,6 +753,7 @@ def init_admin_client(self, verify_connection: bool = True) -> None:
             try:
                 self.admin_client = KafkaAdminClient(
                     bootstrap_servers=self.config.bootstrap_uri,
+                    client_id=self.config.client_id,
                     security_protocol=self.config.security_protocol,
                     ssl_cafile=self.config.ssl_cafile,
                     ssl_certfile=self.config.ssl_certfile,