<!-- Love localstack? Please consider supporting our collective:
👉  https://opencollective.com/localstack/donate -->

# Type of request: This is a ...

[ X ] bug report
[ ] feature request

# Detailed description

On BatchWriteItem if items are throttled, we are expected to receive a list of `UnprocessedItems` in the response instead of HTTP 500

## Expected behavior

Receive HTTP 200 but with Unprocessed Items in the response: https://docs.aws.amazon.com/amazondynamodb/latest/APIReference/API_BatchWriteItem.html#DDB-BatchWriteItem-response-UnprocessedItems

## Actual behavior

Receive HTTP 500 ProvisionedThroughPutException. "All or nothing"

# Steps to reproduce

Start LocalStack with DynamoDb Service and "DYNAMODB_ERROR_PROBABILITY=1.0"

## Command used to start LocalStack

```
new LocalStackContainer(DockerImageName.parse("localstack/localstack:0.12.9"))
      .withEnv("DYNAMODB_ERROR_PROBABILITY", "1.0")
      .withServices(DYNAMODB)
```

## Client code (AWS SDK code snippet, or sequence of "awslocal" commands)

You can just Randomly generate a big list of objects (exceeding 1kb per batch) and check this behaviour


result = {}
    if table_name not in TABLE_DEFINITIONS:
        LOGGER.warning('Unknown table: %s not found in %s' % (table_name, TABLE_DEFINITIONS))
        return None
    for key in TABLE_DEFINITIONS[table_name]['KeySchema']:
        attr_name = key['AttributeName']
        if attr_name not in item:
            return error_response(
                error_type='ValidationException',
                message='One of the required keys was not given a value'
            )
        result[attr_name] = item[attr_name]
    return result
def dynamodb_get_table_stream_specification(table_name):
    try:
        return get_table_schema(table_name)['Table'].get('StreamSpecification')
    except Exception as e:
        LOGGER.info('Unable to get stream specification for table %s : %s %s' % (table_name, e,
            traceback.format_exc()))
        raise e
def is_kinesis_stream_exists(stream_arn):
    
    kinesis = aws_stack.connect_to_service('kinesis')
    stream_name_from_arn = stream_arn.split('/', 1)[1]
    
    filtered = list(filter(lambda stream_name: stream_name == stream_name_from_arn,
        kinesis.list_streams()['StreamNames']))
    return bool(filtered)
def dynamodb_enable_kinesis_streaming_destination(data, table_def):