Restart app before granting credentials (#46)

src/provider.py

@@ -7,7 +7,7 @@
 import logging
 
 from charms.data_platform_libs.v0.data_interfaces import KafkaProvides, TopicRequestedEvent
-from ops.charm import RelationBrokenEvent
+from ops.charm import RelationBrokenEvent, RelationCreatedEvent
 from ops.framework import Object
 
 from auth import KafkaAuth
@@ -34,10 +34,17 @@ def __init__(self, charm) -> None:
 
         self.kafka_provider = KafkaProvides(self.charm, REL_NAME)
 
+        self.framework.observe(self.charm.on[REL_NAME].relation_created, self._on_relation_created)
+
         self.framework.observe(self.charm.on[REL_NAME].relation_broken, self._on_relation_broken)
 
         self.framework.observe(self.kafka_provider.on.topic_requested, self.on_topic_requested)
 
+    def _on_relation_created(self, event: RelationCreatedEvent) -> None:
+        """Handler for `kafka-client-relation-created` event."""
+        # this will trigger kafka restart (if needed) before granting credentials
+        self.charm._on_config_changed(event)
+
     def on_topic_requested(self, event: TopicRequestedEvent):
         """Handle the on topic requested event."""
         if not self.charm.unit.is_leader():
@@ -53,6 +60,8 @@ def on_topic_requested(self, event: TopicRequestedEvent):
             event.defer()
             return
 
+        self.charm._on_config_changed(event)
+
         extra_user_roles = event.extra_user_roles
         topic = event.topic
 
@@ -69,11 +78,15 @@ def on_topic_requested(self, event: TopicRequestedEvent):
         consumer_group_prefix = (
             event.consumer_group_prefix or f"{username}-" if "consumer" in extra_user_roles else ""
         )
-
-        self.kafka_auth.add_user(
-            username=username,
-            password=password,
-        )
+        try:
+            self.kafka_auth.add_user(
+                username=username,
+                password=password,
+            )
+        except Exception:
+            logger.warning("unable to create user just yet")
+            event.defer()
+            return
 
         # non-leader units need cluster_config_changed event to update their super.users
         self.charm.peer_relation.data[self.charm.app].update({username: password})

tests/unit/test_provider.py

@@ -2,6 +2,7 @@
 # Copyright 2023 Canonical Ltd.
 # See LICENSE file for licensing details.
 
+import io
 import logging
 from pathlib import Path
 from unittest.mock import PropertyMock, patch
@@ -65,6 +66,9 @@ def test_client_relation_created_adds_user(harness):
     harness.add_relation(PEER, CHARM_KEY)
     with (
         patch("charm.KafkaK8sCharm.ready_to_start", new_callable=PropertyMock, return_value=True),
+        patch("ops.model.Container.pull", return_value=io.StringIO("gandalf=grey")),
+        patch("config.KafkaConfig.set_server_properties"),
+        patch("config.KafkaConfig.set_client_properties"),
         patch("auth.KafkaAuth.add_user") as patched_add_user,
         patch("ops.model.Container.exec", return_value=DummyExec()),
         patch(
@@ -95,6 +99,9 @@ def test_client_relation_broken_removes_user(harness):
     with (
         patch("charm.KafkaK8sCharm.ready_to_start", new_callable=PropertyMock, return_value=True),
         patch("auth.KafkaAuth.add_user"),
+        patch("ops.model.Container.pull", return_value=io.StringIO("gandalf=grey")),
+        patch("config.KafkaConfig.set_server_properties"),
+        patch("config.KafkaConfig.set_client_properties"),
         patch("auth.KafkaAuth.delete_user") as patched_delete_user,
         patch("auth.KafkaAuth.remove_all_user_acls") as patched_remove_acls,
         patch("ops.model.Container.exec", return_value=DummyExec()),
@@ -134,6 +141,9 @@ def test_client_relation_joined_sets_necessary_relation_data(harness):
     with (
         patch("charm.KafkaK8sCharm.ready_to_start", new_callable=PropertyMock, return_value=True),
         patch("auth.KafkaAuth.add_user"),
+        patch("ops.model.Container.pull", return_value=io.StringIO("gandalf=grey")),
+        patch("config.KafkaConfig.set_server_properties"),
+        patch("config.KafkaConfig.set_client_properties"),
         patch("ops.model.Container.exec", return_value=DummyExec()),
         patch(
             "config.KafkaConfig.zookeeper_connected", new_callable=PropertyMock, return_value=True