MQTT Client Stops Receiving the Messages From Broker.

MQTT Paho Client stops getting the messages from the broker at some point.

The client is receiving the messages from the broker at the rate of 5000 msg/sec and the message size of each message is around 1 KB.

At broker side, I have seen that client is connected with the broker but somehow broker unable to send messages to the client.

Following is the code I am using to connect broker.

```
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttTopic;

import java.util.*;
import java.util.concurrent.atomic.AtomicLong;

public class Test_client implements MqttCallbackExtended{
public static AtomicLong count = new AtomicLong();

String topic = "mqtt/test/#";
String broker = "tcp://broker-ip:1883";
String clientId = "client-id-1";
MqttClient client;
MqttConnectOptions connOpt;

public void connect(){
try {

System.out.println("Connecting to the broker..");
this.client = new MqttClient(broker, clientId);
this.connOpt = new MqttConnectOptions();

// MQTT Configurations
//this.connOpt.setMqttVersion(4);
this.connOpt.setCleanSession(true);
this.connOpt.setKeepAliveInterval(30);
this.connOpt.setAutomaticReconnect(true);
this.connOpt.setUserName("xyz");
this.connOpt.setPassword("abcd".toCharArray());

// Specifying the callback
this.client.setCallback(this);

// Subscriber
this.client.connect(connOpt);
this.client.subscribe(topic, 1);
System.out.println("Connected to the broker.. ");
}
catch(Exception error){
System.out.println("Exception occured while connecting to broker");
}
}

/**
*
* connectionLost
* This callback is invoked upon losing the MQTT connection.
*
*/
@Override
public void connectionLost(Throwable t) {
System.out.println("Connection lost!");
}

/**
*
* deliveryComplete
* This callback is invoked when a message published by this client
* is successfully received by the broker.
*
*/
@Override
public void deliveryComplete(IMqttDeliveryToken token) {
//System.out.println("Pub complete" + new String(token.getMessage().getPayload()));

}

//@Override
public void connectComplete(boolean reconnect, java.lang.String serverURI){
System.out.println("Re-Connection Attempt " + reconnect);
if(reconnect) {
try {
this.client.subscribe(topic, 1);
} catch (MqttException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}

/**
*
* messageArrived
* This callback is invoked when a message is received on a subscribed topic.
*
*/
@Override
public void messageArrived(String topic, MqttMessage message) throws Exception {
/*
System.out.println("-------------------------------------------------");
System.out.println("| Topic:" + topic);
System.out.println("| Message: " + new String(message.getPayload()));
System.out.println("-------------------------------------------------");
*/

count.incrementAndGet();
}


public static void main(String args[]) throws InterruptedException {
Test_client client_mqtt = new Test_client();
client_mqtt.connect();

while(true){
Date date = new Date();
System.out.println(date.toString() + ": Total Messages Count: " + Test_client.count.get());
Thread.sleep(10000);
}

}

}

```