def subscribe(topic_or_regex, default_offset: nil, start_from_beginning: true, max_bytes_per_partition: 1048576)
      default_offset ||= start_from_beginning ? :earliest : :latest

      if topic_or_regex.is_a?(Regexp)
        cluster_topics.select { |topic| topic =~ topic_or_regex }.each do |topic|
          subscribe_to_topic(topic, default_offset, start_from_beginning, max_bytes_per_partition)
        end
      else
        subscribe_to_topic(topic_or_regex, default_offset, start_from_beginning, max_bytes_per_partition)
      end

      nil
    end
def subscribe(topic_or_regex, default_offset: nil, start_from_beginning: true, max_bytes_per_partition: 1048576)
      default_offset ||= start_from_beginning ? :earliest : :latest

      if topic_or_regex.is_a?(Regexp)
        cluster_topics.select { |topic| topic =~ topic_or_regex }.each do |topic|
          subscribe_to_topic(topic, default_offset, start_from_beginning, max_bytes_per_partition)
        end
      else
        subscribe_to_topic(topic_or_regex, default_offset, start_from_beginning, max_bytes_per_partition)
      end

      nil
    end
Subscribes the consumer to a topic.

 Typically you either want to start reading messages from the very
 beginning of the topic's partitions or you simply want to wait for new
 messages to be written. In the former case, set `start_from_beginning`
 to true (the default); in the latter, set it to false.

 @param topic_or_regex [String, Regexp] subscribe to single topic with a string
   or multiple topics matching a regex.
 @param default_offset [Symbol] whether to start from the beginning or the
   end of the topic's partitions. Deprecated.
 @param start_from_beginning [Boolean] whether to start from the beginning
   of the topic or just subscribe to new messages being produced. This
   only applies when first consuming a topic partition – once the consumer
   has checkpointed its progress, it will always resume from the last
   checkpoint.
 @param max_bytes_per_partition [Integer] the maximum amount of data fetched
   from a single partition at a time.
 @return [nil]