<commit_before>import json
import boto.kinesis
from motorway.decorators import batch_process
from motorway.intersection import Intersection
from time import sleep
import logging
logger = logging.getLogger(__name__)


class KinesisInsertIntersection(Intersection):
    stream_name = None

    def __init__(self, **kwargs):
        super(KinesisInsertIntersection, self).__init__(**kwargs)
        self.conn = boto.kinesis.connect_to_region(**self.connection_parameters())
        assert self.stream_name, "Please define attribute stream_name on your KinesisInsertIntersection"

    def connection_parameters(self):
        return {
            'region_name': 'eu-west-1',
            # Add this or use ENV VARS
            # 'aws_access_key_id': '',
            # 'aws_secret_access_key': ''
        }

    @batch_process(limit=500, wait=1)
    def process(self, messages):
        """
        wait 1 second and get up to 500 items 
        Each PutRecords request can support up to 500 records. 
        Each record in the request can be as large as 1 MB, up to a limit of 5 MB for the entire request, including partition keys. 
        Each shard can support writes up to 1,000 records per second, up to a maximum data write total of 1 MB per second.
        This means we can run 2 intersections (2 x 500 records) submitting to the same shard before hitting the write limit (1000 records/sec)
        If we hit the write limit we wait 2 seconds and try to send the records that failed again, rinse and repeat
        If any other error than ProvisionedThroughputExceededException (only InternalFailure is currently supported) is returned in the response
        we log it using loglevel error and dump the message for replayability instead of raising an exception that would drop the whole batch.
        So if you are going to use this intersection in production be sure to monitor and handle the messages with log level error!
        :param messages: 
        :return: 
        """
        records = []
        for message in messages:
            kinesis_record = {'PartitionKey': message.grouping_value, 'Data': json.dumps(message.content)}
            records.append(kinesis_record)
        response = self.conn.put_records(records, self.stream_name)
        failed_records = []
        for i, record in enumerate(response['Records']):
            if len(record.get('ErrorCode', '')) > 0:
                if record['ErrorCode'] != 'ProvisionedThroughputExceededException':
                    s = '%s for message: %s' % (record['ErrorMessage'], json.dumps(messages[i]))
                    logger.error(s)
                else:
                    failed_records.append(messages[i])
            else:
                self.ack(messages[i])

        if len(failed_records) > 0:
            sleep(2)
            self.process(failed_records)

        yield<commit_msg>adjusted wait time for KinesisInsertIntersection<commit_after>import json
import boto.kinesis
from motorway.decorators import batch_process
from motorway.intersection import Intersection
from time import sleep
import logging
logger = logging.getLogger(__name__)


class KinesisInsertIntersection(Intersection):
    stream_name = None

    def __init__(self, **kwargs):
        super(KinesisInsertIntersection, self).__init__(**kwargs)
        self.conn = boto.kinesis.connect_to_region(**self.connection_parameters())
        assert self.stream_name, "Please define attribute stream_name on your KinesisInsertIntersection"

    def connection_parameters(self):
        return {
            'region_name': 'eu-west-1',
            # Add this or use ENV VARS
            # 'aws_access_key_id': '',
            # 'aws_secret_access_key': ''
        }

    @batch_process(limit=500, wait=3)
    def process(self, messages):
        """
        wait 1 second and get up to 500 items 
        Each PutRecords request can support up to 500 records. 
        Each record in the request can be as large as 1 MB, up to a limit of 5 MB for the entire request, including partition keys. 
        Each shard can support writes up to 1,000 records per second, up to a maximum data write total of 1 MB per second.
        This means we can run 2 intersections (2 x 500 records) submitting to the same shard before hitting the write limit (1000 records/sec)
        If we hit the write limit we wait 2 seconds and try to send the records that failed again, rinse and repeat
        If any other error than ProvisionedThroughputExceededException (only InternalFailure is currently supported) is returned in the response
        we log it using loglevel error and dump the message for replayability instead of raising an exception that would drop the whole batch.
        So if you are going to use this intersection in production be sure to monitor and handle the messages with log level error!
        :param messages: 
        :return: 
        """
        records = []
        for message in messages:
            kinesis_record = {'PartitionKey': message.grouping_value, 'Data': json.dumps(message.content)}
            records.append(kinesis_record)
        response = self.conn.put_records(records, self.stream_name)
        failed_records = []
        for i, record in enumerate(response['Records']):
            if len(record.get('ErrorCode', '')) > 0:
                if record['ErrorCode'] != 'ProvisionedThroughputExceededException':
                    s = '%s for message: %s' % (record['ErrorMessage'], json.dumps(messages[i]))
                    logger.error(s)
                    self.fail(messages[i])
                else:
                    logger.warning(record['ErrorCode'])
                    failed_records.append(messages[i])
            else:
                self.ack(messages[i])

        if len(failed_records) > 0:
            sleep(1)
            self.process(failed_records)

        yield
{'fraction_non_alphanumeric': 0.07383463763184757, 'fraction_numerical': 0.010888057162300102, 'mean_word_length': 2.6001224739742805, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, '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': '3947553', 'n_tokens_mistral': 1571, 'n_tokens_neox': 1470, 'n_words': 563}