docs: def _ensure_valid_topic_name(self, topic):
          """ Ensures that the topic name is valid according to the kafka source. """
  
          # See Kafka Source:
          # https://github.com/apache/kafka/blob/39eb31feaeebfb184d98cc5d94da9148c2319d81/clients/src/main/java/org/apache/kafka/common/internals/Topic.java
          if topic is None:
              raise TypeError('All topics must not be None')
          if not isinstance(topic, six.string_types):
              raise TypeError('All topics must be strings')
          if len(topic) == 0:
              raise ValueError('All topics must be non-empty strings')
          if topic == '.' or topic == '..':
              raise ValueError('Topic name cannot be "." or ".."')
          if len(topic) > self._MAX_NAME_LENGTH:
              raise ValueError('Topic name is illegal, it can\'t be longer than {0} characters, topic: "{1}"'.format(self._MAX_NAME_LENGTH, topic))
          if not self._TOPIC_LEGAL_CHARS.match(topic):
              raise ValueError('Topic name "{0}" is illegal, it contains a character other than ASCII alphanumerics, ".", "_" and "-"'.format(topic))