[DPE-2242] Add kafka upgrade

lib/charms/data_platform_libs/v0/upgrade.py

@@ -284,7 +284,7 @@ def restart(self, event) -> None:
 
 # Increment this PATCH version before using `charmcraft publish-lib` or reset
 # to 0 if you are raising the major API version
-LIBPATCH = 9
+LIBPATCH = 10
 
 PYDEPS = ["pydantic>=1.10,<2"]
 
@@ -817,7 +817,7 @@ def idle(self) -> Optional[bool]:
         Returns:
             True if all application units in idle state. Otherwise False
         """
-        return self.cluster_state == "idle"
+        return set(self.unit_states) == {"idle"}
 
     @abstractmethod
     def pre_upgrade_check(self) -> None:

src/config.py

@@ -453,6 +453,18 @@ def log_dirs(self) -> str:
             [os.fspath(storage.location) for storage in self.charm.model.storages["data"]]
         )
 
+    @property
+    def inter_broker_protocol_version(self) -> str:
+        """Creates the protocol version from the kafka version.
+
+        Returns:
+            string with the `major.minor` version
+        """
+        # Remove patch number from full vervion.
+        major_minor = self.charm.upgrade.current_version.split(".", maxsplit=2)
+        major_minor.pop()
+        return ".".join(major_minor)
+
     @property
     def rack_properties(self) -> List[str]:
         """Builds all properties related to rack awareness configuration.
@@ -512,7 +524,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"inter.broker.protocol.version={self.charm.upgrade.current_version}",
+                f"inter.broker.protocol.version={self.inter_broker_protocol_version}",
             ]
             + self.config_properties
             + self.scram_properties

tests/unit/test_config.py

@@ -13,6 +13,7 @@
 from literals import (
     ADMIN_USER,
     CHARM_KEY,
+    DEPENDENCIES,
     INTER_BROKER_USER,
     INTERNAL_USERS,
     JMX_EXPORTER_PORT,
@@ -402,6 +403,27 @@ def test_rack_properties(harness: Harness):
         assert "broker.rack=gondor-west" in harness.charm.kafka_config.server_properties
 
 
+def test_inter_broker_protocol_version(harness):
+    """Checks that rack properties are added to server properties."""
+    harness.add_relation(PEER, CHARM_KEY)
+    zk_relation_id = harness.add_relation(ZK, CHARM_KEY)
+    harness.update_relation_data(
+        zk_relation_id,
+        harness.charm.app.name,
+        {
+            "chroot": "/kafka",
+            "username": "moria",
+            "password": "mellon",
+            "endpoints": "1.1.1.1,2.2.2.2",
+            "uris": "1.1.1.1:2181/kafka,2.2.2.2:2181/kafka",
+            "tls": "disabled",
+        },
+    )
+    assert len(DEPENDENCIES["kafka_service"]["version"].split(".")) == 3
+
+    assert f"inter.broker.protocol.version=3.3" in harness.charm.kafka_config.server_properties
+
+
 def test_super_users(harness):
     """Checks super-users property is updated for new admin clients."""
     peer_relation_id = harness.add_relation(PEER, CHARM_KEY)