On subscription you didn't get all retained messages

- [x] Bug exists Release Version 1.2.5 ( Master Branch)
- [x] Bug exists in MQTTv3 Client on Snapshot Version 1.2.6-SNAPSHOT (Develop Branch)
- [x] Bug exists in MQTTv5 Client on Snapshot Version 1.2.6-SNAPSHOT (Develop Branch)

We have a broker (Mosquitto 2.0.18) with lots of retained messages in several topic. A new connected paho client with several wildcard subscriptions didn't get all the retained messages. This miss behavior depends on message and topic count, topic lengths and message size.

I have created a test to reproduce the issue:
```
/*
* Copyright (c) 2012 - 2017 Data In Motion and others.
* All rights reserved.
*
* This program and the accompanying materials are made available under the terms of the
* Eclipse Public License v1.0 which accompanies this distribution, and is available at
* http://www.eclipse.org/legal/epl-v10.html
*
* Contributors:
* Data In Motion
*******************************************************************************/

package org.eclipse.paho.client.mqttv3.test;

import static org.junit.Assert.assertTrue;

import java.net.URI;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;

import org.eclipse.paho.client.mqttv3.IMqttClient;
import org.eclipse.paho.client.mqttv3.IMqttMessageListener;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttPersistenceException;
import org.eclipse.paho.client.mqttv3.test.client.MqttClientFactoryPaho;
import org.eclipse.paho.client.mqttv3.test.logging.LoggingUtilities;
import org.eclipse.paho.client.mqttv3.test.properties.TestProperties;
import org.eclipse.paho.client.mqttv3.test.utilities.Utility;
import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.BeforeClass;
import org.junit.Test;

/**
* Tests shows an issue with a wildcard subscription to a topic with lots of
* retained messages.
*/

public class RetainedTest {

static final Class cclass = RetainedTest.class;
private static final String className = cclass.getName();
private static final Logger log = Logger.getLogger(className);

private static final int MESSAGE_COUNT = 25000;
private static final int TOPIC_COUNT = 4;
private static URI serverURI;
private static MqttClientFactoryPaho clientFactory;
private static String topicPrefix;

/**
* @throws Exception
*/
@BeforeClass
public static void setUpBeforeClass() throws Exception {

try {
String methodName = Utility.getMethodName();
LoggingUtilities.banner(log, cclass, methodName);

serverURI = TestProperties.getServerURI();
clientFactory = new MqttClientFactoryPaho();
clientFactory.open();
topicPrefix = "RetainedTest-" + UUID.randomUUID().toString() + "-";

} catch (Exception exception) {
log.log(Level.SEVERE, "caught exception:", exception);
throw exception;
}
}

/**
* @throws Exception
*/
@AfterClass
public static void tearDownAfterClass() throws Exception {
String methodName = Utility.getMethodName();
LoggingUtilities.banner(log, cclass, methodName);

try {
if (clientFactory != null) {
clientFactory.close();
clientFactory.disconnect();
}
} catch (Exception exception) {
log.log(Level.SEVERE, "caught exception:", exception);
}
}

/**
* @throws Exception
*/
@Test
public void testSubscribeRetained() throws Exception {
String methodName = Utility.getMethodName();
LoggingUtilities.banner(log, cclass, methodName);

IMqttClient client = null;
try {
String clientId = methodName + "-pub";
client = clientFactory.createMqttClient(serverURI, clientId);

log.info("Connecting...(serverURI:" + serverURI + ", ClientId:" + clientId);
client.connect();

for (int i = 0; i < TOPIC_COUNT; i++) {
String topic = topicPrefix + i + "/";
publishMessages(client, topic);
}

log.info("Disconnecting... Closing ...");
client.disconnect();
client.close();

log.info("Creating new client ... ");
clientId = methodName + "-sub";
final CountDownLatch messageLatch = new CountDownLatch(TOPIC_COUNT * MESSAGE_COUNT);
client = clientFactory.createMqttClient(serverURI, clientId);

log.info("Connecting...(serverURI:" + serverURI + ", ClientId:" + clientId);
client.connect();

for (int i = 0; i < TOPIC_COUNT; i++) {
final String t = topicPrefix + i + "/#";
log.info("Subscribe to " + t);
client.subscribe(t, 1, new IMqttMessageListener() {
int messageCounter;
@Override
public void messageArrived(String topic, MqttMessage message) throws Exception {
messageLatch.countDown();
if (++messageCounter % 5000 == 0) {
log.info(messageCounter + " messages arrived on " + t);
}
}
});
}
log.info("wait for the " + TOPIC_COUNT * MESSAGE_COUNT + " retained messages to arrive.");
boolean result = messageLatch.await(10, TimeUnit.SECONDS);
assertTrue(messageLatch.getCount() + " messages didn't arrive.", result);

} catch (MqttException exception) {
log.log(Level.SEVERE, "caught exception:", exception);
Assert.fail("Unexpected exception: " + exception);
} finally {
if (client != null) {
log.info("Disconnecting...");
client.disconnect();
log.info("Close...");
client.close();
}
}
}

private void publishMessages(IMqttClient client, String topic) throws MqttException, MqttPersistenceException {
log.info("Publishing " + MESSAGE_COUNT + " retained messages to " + topic + ".");
for (int i = 0; i < MESSAGE_COUNT; i++) {
client.publish(topic + "1234567890-1234567890-1234567890-1234567890-" + i,
("abcdefghijklmnopqrstuvw-abcdefghijklmnopqrstuvw-abcdefghijklmnopqrstuvw-abcdefghijklmnopqrstuvw-abcdefghijklmnopqrstuvw-abcdefghijklmnopqrstuvw-"
+ i).getBytes(),
0, true);
}
}

}
```

Start with the RetainedTest reproduction in the issue and run it against Mosquitto 2.0.18, using the four wildcard subscriptions and 25,000 retained messages per topic. Trace the client's subscription and retained-message handling to determine why messages are missed; done means the latch receives all 100,000 retained messages.

Written by the indexing model from the issue text.