Получение java.lang.OutOfMemoryError: пространство кучи Java с помощью org.eclipse.paho.mqttv5.client.internal.ClientSta ⇐ JAVA

[Anonymous]

Снимок экрана дампа памяти


Когда я запускаю приложение Springboot в течение некоторого времени, я получаю сообщение java.lang.OutOfMemoryError: пространство кучи Java
Похоже, что размер org.eclipse.paho.mqttv5.client.internal.ClientState продолжает увеличиваться.
Я что-то упустил?
** Обратите внимание, что я устанавливаю options.setCleanStart(false); и options.setSessionExpiryInterval(4294967295L);
так что, если экземпляр приложения умирает, то новый экземпляр продолжает работу, как если бы это был тот же клиент, поэтому он потребляет сообщение, которое не было использовано ранее.
public void connect() {
try {
client = new MqttAsyncClient(brokerUrl, "my-client");

MqttConnectionOptions options = new MqttConnectionOptions();
options.setUserName(username);
options.setPassword(password.getBytes());
options.setAutomaticReconnect(true);
options.setCleanStart(false); // Persistent session
options.setSessionExpiryInterval(4294967295L); // Max session expiry

client.setCallback(new MqttCallback() {
@Override
public void messageArrived(String topic, MqttMessage message) {
executor.submit(() -> {
try {
String payload = new String(message.getPayload());
routeMessage(topic, payload);
} catch (Exception e) {
log.error(" Error processing MQTT message from topic {}: {}",
topic, e.getMessage(), e);
}
});
}

@Override
public void disconnected(MqttDisconnectResponse disconnectResponse) {
log.warn(" Disconnected from MQTT broker: {}",
disconnectResponse.getReasonString());
}

@Override
public void mqttErrorOccurred(MqttException exception) {
log.error(" MQTT error occurred", exception);
}

@Override
public void deliveryComplete(IMqttToken token) {
// Not used for subscriptions
}

@Override
public void connectComplete(boolean reconnect, String serverURI) {
log.info(" MQTT connection complete to {}", serverURI);
}

@Override
public void authPacketArrived(int reasonCode, MqttProperties properties) {
// Optional, unused
}
});

client.connect(options).waitForCompletion();

log.info(" Connected to MQTT broker at {}", brokerUrl);

for (String topic : topics) {
client.subscribe(topic, 2).waitForCompletion();
log.info(" Subscribed to topic: {}", topic);
}

} catch (MqttException e) {
log.error(" Failed to connect or subscribe to MQTT broker", e);
}
}


Подробнее здесь: https://stackoverflow.com/questions/797 ... aho-mqttv5