public function flush($untilEmpty = true)
    {
        // Send BatchWriteItem requests until the queue is empty
        $keepFlushing = true;
        while ($this->queue && $keepFlushing) {
            $commands = $this->prepareCommands();
            $pool = new CommandPool($this->client, $commands, [
                'before' => $this->config['before'],
                'concurrency' => $this->config['pool_size'],
                'fulfilled'   => function (ResultInterface $result) {
                    // Re-queue any unprocessed items
                    if ($result->hasKey('UnprocessedItems')) {
                        $this->retryUnprocessed($result['UnprocessedItems']);
                    }
                },
                'rejected' => function ($reason) {
                    if ($reason instanceof AwsException) {
                        $code = $reason->getAwsErrorCode();
                        if ($code === 'ProvisionedThroughputExceededException') {
                            $this->retryUnprocessed($reason->getCommand()['RequestItems']);
                        } elseif (is_callable($this->config['error'])) {
                            $this->config['error']($reason);
                        }
                    }
                }
            ]);
            $pool->promise()->wait();
            $keepFlushing = (bool) $untilEmpty;
        }

        return $this;
    }