<commit_before>pub use super::*;
use kafka::producer::{Producer, RequiredAcks};
use kafka;
use kafka::consumer::Consumer;
use std::time::Duration;

pub fn test_producer() -> Producer {
    Producer::from_hosts(vec![LOCAL_KAFKA_BOOTSTRAP_HOST.to_owned()])
        .with_ack_timeout(Duration::from_secs(1))
        .with_required_acks(RequiredAcks::One)
        .create()
        .unwrap()
}

/// Return a Consumer builder with some defaults
pub fn test_consumer_builder() -> kafka::consumer::Builder {
    Consumer::from_hosts(vec![LOCAL_KAFKA_BOOTSTRAP_HOST.to_owned()])
        .with_topic_partitions(TEST_TOPIC_NAME.to_owned(), &TEST_TOPIC_PARTITIONS)
        .with_group(TEST_GROUP_NAME.to_owned())
        .with_fallback_offset(kafka::consumer::FetchOffset::Latest)
        .with_offset_storage(kafka::consumer::GroupOffsetStorage::Kafka)
}

/// Return a ready Kafka consumer with all default settings
pub fn test_consumer() -> Consumer {
    test_consumer_builder().create().unwrap()
}

mod producer;
mod consumer;
<commit_msg>Make sure consumer is always at the latest offset<commit_after>pub use super::*;
use kafka::producer::{Producer, RequiredAcks};
use kafka;
use kafka::client::FetchOffset;
use kafka::consumer::Consumer;
use std::time::Duration;

pub fn test_producer() -> Producer {
    Producer::from_hosts(vec![LOCAL_KAFKA_BOOTSTRAP_HOST.to_owned()])
        .with_ack_timeout(Duration::from_secs(1))
        .with_required_acks(RequiredAcks::All)
        .create()
        .unwrap()
}

/// Return a Consumer builder with some defaults
pub fn test_consumer_builder() -> kafka::consumer::Builder {
    Consumer::from_hosts(vec![LOCAL_KAFKA_BOOTSTRAP_HOST.to_owned()])
        .with_topic_partitions(TEST_TOPIC_NAME.to_owned(), &TEST_TOPIC_PARTITIONS)
        .with_group(TEST_GROUP_NAME.to_owned())
        .with_fallback_offset(kafka::consumer::FetchOffset::Latest)
        .with_offset_storage(kafka::consumer::GroupOffsetStorage::Kafka)
}

/// Return a ready Kafka consumer with all default settings
pub fn test_consumer() -> Consumer {
    let mut consumer = test_consumer_builder().create().unwrap();
    let topics = [TEST_TOPIC_NAME, TEST_TOPIC_NAME_2];

    // Fetch the latest offsets and commit those so that this consumer
    // is always at the latest offset before being used.
    {
        let client = consumer.client_mut();
        let latest_offsets = client.fetch_offsets(&topics, FetchOffset::Latest).unwrap();

        for (topic, partition_offsets) in latest_offsets {
            for po in partition_offsets {
                client
                    .commit_offset(TEST_GROUP_NAME, &topic, po.partition, po.offset)
                    .unwrap();
            }
        }
    }

    consumer
}

mod producer;
mod consumer;

{'fraction_non_alphanumeric': 0.11734878667149584, 'fraction_numerical': 0.0010865628395508873, 'mean_word_length': 3.897163120567376, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 9, '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': '5925679', 'n_tokens_mistral': 882, 'n_tokens_neox': 854, 'n_words': 172}