Always apply whole config dict

esque/controller/topic_controller.py

@@ -96,10 +96,22 @@ def create_topics(self, topics: List[Topic]):
     @invalidate_cache_after
     def alter_configs(self, topics: List[Topic]):
         for topic in topics:
-            config_resource = ConfigResource(ConfigResource.Type.TOPIC, topic.name, topic.config)
+            altered_config = self._get_altered_config(topic)
+            config_resource = ConfigResource(ConfigResource.Type.TOPIC, topic.name, altered_config)
             future_list = self.cluster.confluent_client.alter_configs([config_resource])
             ensure_kafka_future_done(next(islice(future_list.values(), 1)))
 
+    def _get_altered_config(self, topic: Topic) -> Dict[str, str]:
+        cluster_topic = self.get_cluster_topic(topic.name)
+        current_config = cluster_topic.config.items()
+        altered_config = {}
+        for name, value in current_config:
+            if name in topic.config:
+                altered_config[name] = topic.config[name]
+                continue
+            altered_config[name] = value
+        return altered_config
+
     @invalidate_cache_after
     def delete_topic(self, topic: Topic):
         future = self.cluster.confluent_client.delete_topics([topic.name])[topic.name]
@@ -128,7 +140,7 @@ def get_offsets_closest_to_timestamp(
             topic_partition.partition: topic_partition.offset for topic_partition in topic_partitions_with_new_offsets
         }
 
-    def update_from_cluster(self, topic: Topic):
+    def update_from_cluster(self, topic: Topic) -> Topic:
         """Takes a topic and, based on its name, updates all attributes from the cluster"""
 
         confluent_topic: ConfluentTopic = self._get_client_topic(topic.name, ClientTypes.Confluent)

tests/integration/test_topic_controller.py

@@ -55,6 +55,25 @@ def test_alter_topic_config_works(topic_controller: TopicController, topic_id: s
     assert final_config.get("cleanup.policy") == "compact"
 
 
+@pytest.mark.integration
+def test_alter_topic_config_only_changes_mentioned_attributes(topic_controller: TopicController, topic_id: str):
+    initial_topic = Topic(topic_id, config={"cleanup.policy": "delete", "min.compaction.lag": "1000000"})
+
+    topic_controller.create_topics([initial_topic])
+    topic_controller.update_from_cluster(initial_topic)
+    config = initial_topic.config
+    assert config.get("cleanup.policy") == "delete"
+    assert config.get("min.compaction.lag") == "1000000"
+    change_topic = Topic(topic_id, config={"cleanup.policy": "compact"})
+    topic_controller.alter_configs([change_topic])
+    topic_controller.update_from_cluster(change_topic)
+    after_changes_applied_topic = topic_controller.get_cluster_topic(topic_id)
+
+    final_config = after_changes_applied_topic.config
+    assert final_config.get("cleanup.policy") == "compact"
+    assert final_config.get("min.compaction.lag") == "1000000"
+
+
 @pytest.mark.integration
 def test_topic_listing_works(topic_controller: TopicController, topic: str):
     topics = topic_controller.list_topics()