def clear_offsets_excluding(excluded)
      # Clear all offsets that aren't in `excluded`.
      @processed_offsets.each do |topic, partitions|
        partitions.keep_if do |partition, _|
          excluded.fetch(topic, []).include?(partition)
        end
      end

      # Clear the cached commits from the brokers.
      @committed_offsets = nil
      @resolved_offsets.clear
    end
def clear_offsets_excluding(excluded)
      # Clear all offsets that aren't in `excluded`.
      @processed_offsets.each do |topic, partitions|
        partitions.keep_if do |partition, _|
          excluded.fetch(topic, []).include?(partition)
        end
      end

      # Clear the cached commits from the brokers.
      @committed_offsets = nil
      @resolved_offsets.clear
    end
Clear stored offset information for all partitions except those specified
 in `excluded`.

     offset_manager.clear_offsets_excluding("my-topic" => [1, 2, 3])

 @return [nil]