<commit_before>"""
Primary entrypoint for applications wishing to implement Python Kafka Streams

"""

import logging

import confluent_kafka as kafka

log = logging.getLogger(__name__)

class KafkaStream(object):
    """
    Encapsulates stream graph processing units

    """

    def __init__(self, topology, kafka_config):
        self.topology = topology
        self.kafka_config = kafka_config

        self.consumer = None

    def start(self):
        """
        Begin streaming the data across the topology

        """
        self.consumer = kafka.Consumer({'bootstrap.servers': self.kafka_config.BOOTSTRAP_SERVERS, 'group.id': 'test'})

        #, 'group.id': 'testgroup',
        #'default.topic.config': {'auto.offset.reset': 'smallest'}})
        log.debug('Subscribing to topics %s', self.topology.kafka_topics)
        self.consumer.subscribe(self.topology.kafka_topics)
        log.debug('Subscribed to topics')

        self.run()


    def run(self):
        running = True
        while running:
            msg = self.consumer.poll()
            if not msg.error():
                print('Received message: %s' % msg.value().decode('utf-8'))
            elif msg.error().code() != kafka.KafkaError._PARTITION_EOF:
                print(msg.error())
                running = False
        self.consumer.close()


    <commit_msg>Handle None returned from poll()<commit_after>"""
Primary entrypoint for applications wishing to implement Python Kafka Streams

"""

import logging

import confluent_kafka as kafka

log = logging.getLogger(__name__)

class KafkaStream(object):
    """
    Encapsulates stream graph processing units

    """

    def __init__(self, topology, kafka_config):
        self.topology = topology
        self.kafka_config = kafka_config

        self.consumer = None

    def start(self):
        """
        Begin streaming the data across the topology

        """
        self.consumer = kafka.Consumer({'bootstrap.servers': self.kafka_config.BOOTSTRAP_SERVERS, 'group.id': 'test'})

        #, 'group.id': 'testgroup',
        #'default.topic.config': {'auto.offset.reset': 'smallest'}})
        log.debug('Subscribing to topics %s', self.topology.kafka_topics)
        self.consumer.subscribe(self.topology.kafka_topics)
        log.debug('Subscribed to topics')

        self.run()


    def run(self):
        running = True
        while running:
            msg = self.consumer.poll()
            if msg is None:
                continue
            elif not msg.error():
                print('Received message: %s' % msg.value().decode('utf-8'))
            elif msg.error().code() != kafka.KafkaError._PARTITION_EOF:
                print(msg.error())
                running = False
        self.consumer.close()


    
{'fraction_non_alphanumeric': 0.11762589928057554, 'fraction_numerical': 0.0007194244604316547, 'mean_word_length': 2.350602409638554, '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': '7046202', 'n_tokens_mistral': 821, 'n_tokens_neox': 771, 'n_words': 183}