protected function getMessages($shard_id = NULL, $last_sequence_number = NULL): array
    {

        $records = [];

        // Get the initial iterator.
        try {
            $shard_iterator = $this->getInitialShardIterator($shard_id, $last_sequence_number);
        } catch (\Exception $e) {
            $this->processError($e, [
                'shard_id' => $shard_id,
                'last_sequence_number' => $last_sequence_number,
            ]);

            return $records;
        }

        do {

            $has_records = FALSE;

            // Load batch of records.
            try {
                $res = $this->getClient()->getRecords([
                    'Limit' => $this->batchSize,
                    'ShardIterator' => $shard_iterator,
                    'StreamName' => $this->getStreamName(),
                ]);

                // Get the Shard iterator for next batch.
                $shard_iterator = $res->get('NextShardIterator');
                $behind_latest = $res->get('MillisBehindLatest');

                foreach ($res->search('Records[].[SequenceNumber, Data]') as $event) {
                    list($sequence_number, $message_data) = $event;

                    try {
                        $records[$sequence_number] = $this->getMessageFactory()->unserialize($message_data);
                    } catch (\Exception $e) {
                        $this->processError($e, $event);
                    }

                    $has_records = TRUE;
                }
            } catch (\Exception $e) {
                $this->processError($e);
            }

        } while (!empty($shard_iterator) && !empty($behind_latest) && !$has_records);

        if (!empty($sequence_number)) {
            $this->lastSequenceNumber = $sequence_number;
        }

        if (!empty($behind_latest)) {
            $this->lag = $behind_latest;
        }

        return $records;
    }