<commit_before>from contextlib import contextmanager
import mock
import time
import unittest2
from uuid import uuid4

from pykafka import KafkaClient
from pykafka.simpleconsumer import OwnedPartition, OffsetType
from pykafka.test.utils import get_cluster, stop_cluster
from pykafka.utils.compat import range, iteritems


class TestSimpleConsumer(unittest2.TestCase):
    maxDiff = None
    USE_RDKAFKA = False

    @classmethod
    def setUpClass(cls):
        cls.kafka = get_cluster()
        cls.topic_name = uuid4().hex.encode()
        cls.kafka.create_topic(cls.topic_name, 3, 2)

        # It turns out that the underlying producer used by KafkaInstance will
        # write all messages in a batch to a single partition, though not the
        # same partition every time.  We try to attain some spread here by
        # sending more than one batch:
        batch = 300
        cls.total_msgs = 3 * batch
        cls.client = KafkaClient(cls.kafka.brokers)
        cls.prod = cls.client.topics[cls.topic_name].get_producer(
            min_queued_messages=1
        )
        for _ in range(3):
            for i in range(batch):
                cls.prod.produce('msg {i}'.format(i=i).encode())

        cls.client = KafkaClient(cls.kafka.brokers)

    @classmethod
    def tearDownClass(cls):
        stop_cluster(cls.kafka)

    @contextmanager
    def _get_simple_consumer(self, **kwargs):<commit_msg>test_simpleconsumer: toss out multi-batch producing<commit_after>from contextlib import contextmanager
import mock
import time
import unittest2
from uuid import uuid4

from pykafka import KafkaClient
from pykafka.simpleconsumer import OwnedPartition, OffsetType
from pykafka.test.utils import get_cluster, stop_cluster
from pykafka.utils.compat import range, iteritems


class TestSimpleConsumer(unittest2.TestCase):
    maxDiff = None
    USE_RDKAFKA = False

    @classmethod
    def setUpClass(cls):
        cls.kafka = get_cluster()
        cls.topic_name = uuid4().hex.encode()
        cls.kafka.create_topic(cls.topic_name, 3, 2)

        cls.total_msgs = 1000
        cls.client = KafkaClient(cls.kafka.brokers)
        cls.prod = cls.client.topics[cls.topic_name].get_producer(
            min_queued_messages=1
        )
        for i in range(cls.total_msgs):
            cls.prod.produce('msg {i}'.format(i=i).encode())

        cls.client = KafkaClient(cls.kafka.brokers)

    @classmethod
    def tearDownClass(cls):
        stop_cluster(cls.kafka)

    @contextmanager
    def _get_simple_consumer(self, **kwargs):
{'fraction_non_alphanumeric': 0.08228346456692913, 'fraction_numerical': 0.00905511811023622, 'mean_word_length': 3.0526315789473686, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '3088382', 'n_tokens_mistral': 798, 'n_tokens_neox': 770, 'n_words': 194}