Find a consistent, non-surprising way to handle replication factor and partition amount for new and existing topics #44
Conversation
|
This PR is a bigger than intended, since I've had to refactor the entire Diff logic to make a consistent check possible, and I ran in some unexpected difficulties (ex. the config values we get from the cluster are always strings, but for local topics we can provide values as either string or integers). We now have a TopicDiff class with some more utility functions to keep the validation logic etc. in one place. This should also make integrating #45 into |
|
Weird failure in the travis tests that doesn't show up on local tests...
How the F is it possible to have a replication factor of 0 on a remote topic o_O |
…l-digital/esque into 43-replication-and-partition-numbers
…l-digital/esque into 43-replication-and-partition-numbers
…l-digital/esque into 43-replication-and-partition-numbers
Already ran black. Should be fine now. |
|
|
||
| # TODO: this should be handled correctly by checking the cluster defaults, like | ||
| # if remote == cluster_default(name) and local is None: return | ||
| # currently, if an attribute that was set get's un-set, it's ignored |
There was a problem hiding this comment.
| # currently, if an attribute that was set get's un-set, it's ignored | |
| # currently, if an attribute is set remotely but supposed to be unset now, that won't happen because it's ignored |
| @classmethod | ||
| def from_dict(cls, diff_dict: Dict[str, AttributeDiff]) -> "TopicDiff": | ||
| td = TopicDiff() | ||
| td._diffs = diff_dict |
There was a problem hiding this comment.
| td._diffs = diff_dict | |
| td._diffs = diff_dict.copy() |
| if self.is_only_local: | ||
| return self.__replication_factor | ||
| partition_replication_factors = set(len(p.partition_replicas) for p in self.partitions) | ||
| assert len(partition_replication_factors) == 1, "Different replication factors for partitions!" |
There was a problem hiding this comment.
That might actually happen during partition reassignment... We could either try to find the most occurring value or tell the user to try again later.
| PyKafka = "PyKafka" | ||
|
|
||
|
|
||
| ClientType = Union[ConfluentTopic, PyKafkaTopic] |
There was a problem hiding this comment.
Maybe ClientTopicType? This one is easy to confuse with ClientTypes.
| elif client_type == ClientTypes.PyKafka: | ||
| # at least PyKafka does it's own caching, so we don't have to bother | ||
| pykafka_topics = self.cluster.pykafka_client.cluster.topics | ||
| # PyKafka will try to auto-create the topic if you use `topic[random_name]` |
There was a problem hiding this comment.
| # PyKafka will try to auto-create the topic if you use `topic[random_name]` | |
| # PyKafka will try to auto-create the topic if you use `topic[topic_name]`, so we check first if it exists in `confluent_topics`, we can't check if it exists in `pykafka_topics` since that one lazily evaluated and won't contain it. |
Closes #43