<commit_before>describe Kafka::OffsetManager do
  let(:cluster) { double(:cluster) }
  let(:group) { double(:group) }
  let(:logger) { LOGGER }

  let(:offset_manager) {
    Kafka::OffsetManager.new(
      cluster: cluster,
      group: group,
      logger: logger,
      commit_interval: 0,
      commit_threshold: 0,
    )
  }

  before do
    allow(group).to receive(:commit_offsets)
  end

  describe "#commit_offsets" do
    it "commits the processed offsets to the group" do
      offset_manager.mark_as_processed("greetings", 0, 42)
      offset_manager.mark_as_processed("greetings", 1, 13)

      offset_manager.commit_offsets

      expected_offsets = {
        "greetings" => {
          0 => 43,
          1 => 14,
        }
      }

      expect(group).to have_received(:commit_offsets).with(expected_offsets)
    end
  end

  describe "#next_offset_for" do
    let(:fetched_offsets) { double(:fetched_offsets) }

    before do
      allow(group).to receive(:fetch_offsets).and_return(fetched_offsets)
    end

    it "returns the last committed offset" do
      allow(fetched_offsets).to receive(:offset_for).with("greetings", 0) { 41 }

      offset = offset_manager.next_offset_for("greetings", 0)

      expect(offset).to eq 41
    end

    it "returns the default offset if none have been committed" do
      allow(group).to receive(:assigned_partitions) { { "greetings" => [0] } }
      allow(fetched_offsets).to receive(:offset_for).with("greetings", 0) { -1 }
      allow(cluster).to receive(:resolve_offsets).with("greetings", [0], :latest) { { 0 => 42 } }
      offset_manager.set_default_offset("greetings", :latest)

      offset = offset_manager.next_offset_for("greetings", 0)

      expect(offset).to eq 42
    end
  end

  describe "#clear_offsets_excluding" do
    it "clears offsets except for the partitions in the exclusion list" do
      offset_manager.mark_as_processed("x", 0, 42)
      offset_manager.mark_as_processed("x", 1, 13)

      offset_manager.clear_offsets_excluding("x" => [0])
      offset_manager.commit_offsets

      expected_offsets = {
        "x" => {
          0 => kind_of(Numeric),
        }
      }

      expect(group).to have_received(:commit_offsets).with(expected_offsets)
    end
  end
end<commit_msg>Use explicit value for clear_offsets_excluding test<commit_after>describe Kafka::OffsetManager do
  let(:cluster) { double(:cluster) }
  let(:group) { double(:group) }
  let(:logger) { LOGGER }

  let(:offset_manager) {
    Kafka::OffsetManager.new(
      cluster: cluster,
      group: group,
      logger: logger,
      commit_interval: 0,
      commit_threshold: 0,
    )
  }

  before do
    allow(group).to receive(:commit_offsets)
  end

  describe "#commit_offsets" do
    it "commits the processed offsets to the group" do
      offset_manager.mark_as_processed("greetings", 0, 42)
      offset_manager.mark_as_processed("greetings", 1, 13)

      offset_manager.commit_offsets

      expected_offsets = {
        "greetings" => {
          0 => 43,
          1 => 14,
        }
      }

      expect(group).to have_received(:commit_offsets).with(expected_offsets)
    end
  end

  describe "#next_offset_for" do
    let(:fetched_offsets) { double(:fetched_offsets) }

    before do
      allow(group).to receive(:fetch_offsets).and_return(fetched_offsets)
    end

    it "returns the last committed offset" do
      allow(fetched_offsets).to receive(:offset_for).with("greetings", 0) { 41 }

      offset = offset_manager.next_offset_for("greetings", 0)

      expect(offset).to eq 41
    end

    it "returns the default offset if none have been committed" do
      allow(group).to receive(:assigned_partitions) { { "greetings" => [0] } }
      allow(fetched_offsets).to receive(:offset_for).with("greetings", 0) { -1 }
      allow(cluster).to receive(:resolve_offsets).with("greetings", [0], :latest) { { 0 => 42 } }
      offset_manager.set_default_offset("greetings", :latest)

      offset = offset_manager.next_offset_for("greetings", 0)

      expect(offset).to eq 42
    end
  end

  describe "#clear_offsets_excluding" do
    it "clears offsets except for the partitions in the exclusion list" do
      offset_manager.mark_as_processed("x", 0, 42)
      offset_manager.mark_as_processed("x", 1, 13)

      offset_manager.clear_offsets_excluding("x" => [0])
      offset_manager.commit_offsets

      expected_offsets = {
        "x" => {
          0 => 43,
        }
      }

      expect(group).to have_received(:commit_offsets).with(expected_offsets)
    end
  end
end
{'fraction_non_alphanumeric': 0.11638120333772507, 'fraction_numerical': 0.017127799736495388, 'mean_word_length': 3.1110108303249095, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 19, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3192121', 'n_tokens_mistral': 1580, 'n_tokens_neox': 1444, 'n_words': 306}