Repository: boto/boto3
Issue: Dynamodb batch writer periodically failing

I'm using the following logic to pull content from a list and insert it into DynamoDB

```
dynamodb = boto3.resource("dynamodb")
keys_table = dynamodb.Table("my-dynamodb-table")

with keys_table.batch_writer() as batch:
        for key in objects[tmp_id]:
            batch.put_item(Item={
                    "cluster": cluster,
                    "tmp_id": tmp_id,
                    "manifest": manifest_key,
                    "key": key,
                    "timestamp": timestamp
            })
```

It appears to periodically append more than the 25 item limit to the batch and thus fails with the following error:

```
  File "/Library/Python/2.7/site-packages/boto3/dynamodb/table.py", line 121, in __exit__
    self._flush()
  File "/Library/Python/2.7/site-packages/boto3/dynamodb/table.py", line 102, in _flush
    RequestItems={self._table_name: self._items_buffer})
  File "/Library/Python/2.7/site-packages/botocore/client.py", line 310, in _api_call
    return self._make_api_call(operation_name, kwargs)
  File "/Library/Python/2.7/site-packages/botocore/client.py", line 383, in _make_api_call
    api_params, operation_model, context=request_context)
  File "/Library/Python/2.7/site-packages/botocore/client.py", line 425, in _convert_to_request_dict
    api_params, operation_model)
  File "/Library/Python/2.7/site-packages/botocore/validate.py", line 273, in serialize_to_request
    raise ParamValidationError(report=report.generate_report())
botocore.exceptions.ParamValidationError: Parameter validation failed:
Invalid length for parameter RequestItems.my-dynamodb-table, value: 26, valid range: 1-25
```

Package version:

```
Python 2.7.10
boto3==1.2.2
botocore==1.3.12
```


Changed files:
boto3/dynamodb/table.py, tests/unit/dynamodb/test_table.py

diff --git a/boto3/dynamodb/table.py b/boto3/dynamodb/table.py
index 519e2fa2..ef70d7e2 100644
--- a/boto3/dynamodb/table.py
+++ b/boto3/dynamodb/table.py
@@ -98,14 +98,16 @@ class BatchWriter(object):
             self._flush()
 
     def _flush(self):
+        items_to_send = self._items_buffer[:self._flush_amount]
+        self._items_buffer = self._items_buffer[self._flush_amount:]
         response = self._client.batch_write_item(
-            RequestItems={self._table_name: self._items_buffer})
+            RequestItems={self._table_name: items_to_send})
         unprocessed_items = response['UnprocessedItems']
 
         if unprocessed_items and unprocessed_items[self._table_name]:
             # Any unprocessed_items are immediately added to the
             # next batch we send.
-            self._items_buffer = unprocessed_items[self._table_name]
+            self._items_buffer.extend(unprocessed_items[self._table_name])
         else:
             self._items_buffer = []
         logger.debug("Batch write sent %s, unprocessed: %s",
diff --git a/tests/unit/dynamodb/test_table.py b/tests/unit/dynamodb/test_table.py
index fe9ad8eb..127dce27 100644
--- a/tests/unit/dynamodb/test_table.py
+++ b/tests/unit/dynamodb/test_table.py
@@ -166,6 +166,67 @@ class BaseTransformationTest(unittest.TestCase):
             },
         ])
 
+    def test_never_send_more_than_max_batch_size(self):
+        # Suppose the server sends backs a response that indicates that
+        # all the items were unprocessed.
+        self.client.batch_write_item.side_effect = [
+            {
+                'UnprocessedItems': {
+                    self.table_name: [
+                        {'PutRequest': {'Item': {'Hash': 'foo1'}}},
+                        {'PutRequest': {'Item': {'Hash': 'foo2'}}},
+                    ],
+                },
+            },
+            {
+                'UnprocessedItems': {
+                    self.table_name: [
+                        {'PutRequest': {'Item': {'Hash': 'foo2'}}},
+                    ],
+                },
+            },
+            {
+                'UnprocessedItems': {}
+            },
+        ]
+        with BatchWriter(self.table_name, self.client, flush_amount=2) as b:
+            b.put_item(Item={'Hash': 'foo1'})
+            b.put_item(Item={'Hash': 'foo2'})
+            b.put_item(Item={'Hash': 'foo3'})
+
+        # Note how we're never sending more than flush_amount=2.
+        first_batch = {
+            'RequestItems': {
+                self.table_name: [
+                    {'PutRequest': {'Item': {'Hash': 'foo1'}}},
+                    {'PutRequest': {'Item': {'Hash': 'foo2'}}},
+                ]
+            }
+        }
+        # Even when the server sends us unprocessed items of 2 elements,
+        # we'll still only send 2 at a time, in order.
+        second_batch = {
+            'RequestItems': {
+                self.table_name: [
+                    {'PutRequest': {'Item': {'Hash': 'foo1'}}},
+                    {'PutRequest': {'Item': {'Hash': 'foo2'}}},
+                ]
+            }
+        }
+        # And then we still see one more unprocessed item so
+        # we need to send another batch.
+        third_batch = {
+            'RequestItems': {
+                self.table_name: [
+                    {'PutRequest': {'Item': {'Hash': 'foo3'}}},
+                    {'PutRequest': {'Item': {'Hash': 'foo2'}}},
+                ]
+            }
+        }
+        self.assert_batch_write_calls_are([first_batch, second_batch,
+                                           third_batch])
+
+
     def test_repeated_flushing_on_exit(self):
         # We're going to simulate unprocessed_items
         # returning multiple unprocessed items across calls.