[DPE-5218] Enable compatibility with ZK restore feature (#137)

Author: Batalex

src/charm.py

@@ -12,6 +12,8 @@
 from charms.prometheus_k8s.v0.prometheus_scrape import MetricsEndpointProvider
 from charms.rolling_ops.v0.rollingops import RollingOpsManager
 from ops import (
+    ActiveStatus,
+    CollectStatusEvent,
     EventBase,
     StatusBase,
 )
@@ -60,6 +62,7 @@ def __init__(self, *args):
         self.restart = RollingOpsManager(self, relation="restart", callback=self._restart_broker)
 
         self.framework.observe(getattr(self.on, "config_changed"), self._on_roles_changed)
+        self.framework.observe(self.on.collect_app_status, self._on_collect_status)
 
         # peer-cluster events are shared between all roles, so necessary to init here to avoid instantiating multiple times
         self.peer_cluster = PeerClusterEventsHandler(self)
@@ -127,6 +130,23 @@ def _set_status(self, key: Status) -> None:
         getattr(logger, log_level.lower())(status.message)
         self.unit.status = status
 
+    def _on_collect_status(self, event: CollectStatusEvent):
+        ready_to_start = self.state.ready_to_start.value.status
+        event.add_status(ready_to_start)
+
+        if not isinstance(ready_to_start, ActiveStatus):
+            return
+
+        if not self.state.runs_broker:
+            # early return, the next checks only concern the broker
+            return
+
+        if not self.broker.workload.active():
+            event.add_status(Status.BROKER_NOT_RUNNING.value.status)
+
+        if not self.state.zookeeper.broker_active():
+            event.add_status(Status.ZK_NOT_CONNECTED.value.status)
+
 
 if __name__ == "__main__":
     main(KafkaCharm)

src/events/zookeeper.py

@@ -54,6 +54,18 @@ def _on_zookeeper_created(self, _) -> None:
 
     def _on_zookeeper_changed(self, event: RelationChangedEvent) -> None:
         """Handler for `zookeeper_relation_created/joined/changed` events, ensuring internal users get created."""
+        if not self.charm.state.zookeeper.endpoints:
+            # Kafka keeps a meta.properties in every log.dir with a unique ClusterID
+            # this ID is provided by ZK, and removing it on relation-changed allows
+            # re-joining a ZK cluster undergoing backup restoration.
+            self.charm.workload.exec(
+                [
+                    "bash",
+                    "-c",
+                    f"""find {self.charm.workload.paths.data_path} -type f -name meta.properties -delete || true""",
+                ]
+            )
+
         if not self.charm.state.zookeeper.zookeeper_connected:
             self.charm._set_status(Status.ZK_NO_DATA)
             return
@@ -71,7 +83,7 @@ def _on_zookeeper_changed(self, event: RelationChangedEvent) -> None:
             event.defer()
             return
 
-        if not self.charm.state.cluster.internal_user_credentials and self.model.unit.is_leader():
+        if self.model.unit.is_leader():
             # loading the minimum config needed to authenticate to zookeeper
             self.dependent.config_manager.set_zk_jaas_config()
             self.dependent.config_manager.set_server_properties()