From d8cbcd1310c8face04e1d6040d1ea03c92f7208c Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 2 Feb 2017 12:10:44 -0500 Subject: [PATCH] INT-3923: Fix MQTT Reconnect Logic JIRA: https://jira.spring.io/browse/INT-3923 Fixes: #2046 Previously, when connection is lost, the inbound adapter attempted to reconnect on a schedule with a fixed delay. If a connection was again lost, while the schedule is still running, we can end up with another scheduled task running. This is benign aside from the DEBUG log noise because the scheduled task tests the connection before reconnecting. However, if the `recoveryInterval` is short, it could consume CPU. Change the reconnect to be a one-time scheduled task and reschedule if it fails to reconnect. Synchronize all access to the `connected` field. Add a test case with a short recovery interval, before this fix, we see many logs `Attempting reconnect`. --- .../MqttPahoMessageDrivenChannelAdapter.java | 26 +++++++------ .../integration/mqtt/MqttAdapterTests.java | 37 +++++++++++++++++++ 2 files changed, 51 insertions(+), 12 deletions(-) diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java index 00be38c522..07f3beedbb 100644 --- a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java +++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -17,6 +17,7 @@ package org.springframework.integration.mqtt.inbound; import java.util.Arrays; +import java.util.Date; import java.util.concurrent.ScheduledFuture; import org.eclipse.paho.client.mqttv3.IMqttAsyncClient; @@ -138,6 +139,7 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv @Override protected void doStart() { + Assert.state(getTaskScheduler() != null, "A 'taskScheduler' is required"); super.doStart(); try { connectAndSubscribe(); @@ -149,7 +151,7 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv } @Override - protected void doStop() { + protected synchronized void doStop() { cancelReconnect(); super.doStop(); if (this.client != null) { @@ -219,7 +221,7 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv } } - private void connectAndSubscribe() throws MqttException { + private synchronized void connectAndSubscribe() throws MqttException { MqttConnectOptions connectionOptions = this.clientFactory.getConnectionOptions(); this.cleanSession = connectionOptions.isCleanSession(); this.consumerStopAction = this.clientFactory.getConsumerStopAction(); @@ -259,10 +261,6 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv if (this.applicationEventPublisher != null) { this.applicationEventPublisher.publishEvent(new MqttSubscribedEvent(this, message)); } - // cancel() after the publish in case we are on that thread; a send to a QueueChannel would fail. - if (this.reconnectFuture != null) { - cancelReconnect(); - } } } @@ -275,19 +273,23 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv private void scheduleReconnect() { try { - this.reconnectFuture = this.getTaskScheduler().scheduleWithFixedDelay(() -> { + this.reconnectFuture = getTaskScheduler().schedule(() -> { try { if (logger.isDebugEnabled()) { logger.debug("Attempting reconnect"); } - if (!MqttPahoMessageDrivenChannelAdapter.this.connected) { - connectAndSubscribe(); + synchronized (MqttPahoMessageDrivenChannelAdapter.this) { + if (!MqttPahoMessageDrivenChannelAdapter.this.connected) { + connectAndSubscribe(); + MqttPahoMessageDrivenChannelAdapter.this.reconnectFuture = null; + } } } catch (MqttException e) { logger.error("Exception while connecting and subscribing", e); + scheduleReconnect(); } - }, this.recoveryInterval); + }, new Date(System.currentTimeMillis() + this.recoveryInterval)); } catch (Exception e) { logger.error("Failed to schedule reconnect", e); @@ -295,7 +297,7 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv } @Override - public void connectionLost(Throwable cause) { + public synchronized void connectionLost(Throwable cause) { this.logger.error("Lost connection:" + cause.getMessage() + "; retrying..."); this.connected = false; scheduleReconnect(); diff --git a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java index a8b31db709..c4af79a4ee 100644 --- a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java +++ b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java @@ -17,6 +17,7 @@ package org.springframework.integration.mqtt; import static org.hamcrest.Matchers.instanceOf; +import static org.hamcrest.Matchers.lessThanOrEqualTo; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertSame; @@ -24,6 +25,8 @@ import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.BDDMockito.given; +import static org.mockito.BDDMockito.willAnswer; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.mock; @@ -39,11 +42,13 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; import javax.net.SocketFactory; import org.aopalliance.intercept.MethodInterceptor; +import org.apache.commons.logging.Log; import org.eclipse.paho.client.mqttv3.IMqttToken; import org.eclipse.paho.client.mqttv3.MqttAsyncClient; import org.eclipse.paho.client.mqttv3.MqttCallback; @@ -57,6 +62,7 @@ import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence; import org.junit.Test; import org.springframework.aop.framework.ProxyFactoryBean; +import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.BeanFactory; import org.springframework.context.ApplicationEventPublisher; import org.springframework.context.annotation.AnnotationConfigApplicationContext; @@ -77,6 +83,7 @@ import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter; import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; import org.springframework.messaging.support.GenericMessage; +import org.springframework.scheduling.TaskScheduler; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; /** @@ -301,6 +308,7 @@ public class MqttAdapterTests { } assertThat(event, instanceOf(MqttSubscribedEvent.class)); assertEquals("Connected and subscribed to [baz, fix]", ((MqttSubscribedEvent) event).getMessage()); + taskScheduler.destroy(); } @Test @@ -369,6 +377,34 @@ public class MqttAdapterTests { ctx.close(); } + @Test + public void testReconnect() throws Exception { + final MqttAsyncClient client = mock(MqttAsyncClient.class); + MqttPahoMessageDrivenChannelAdapter adapter = buildAdapter(client, null, ConsumerStopAction.UNSUBSCRIBE_NEVER); + adapter.setRecoveryInterval(10); + Log logger = spy(TestUtils.getPropertyValue(adapter, "logger", Log.class)); + new DirectFieldAccessor(adapter).setPropertyValue("logger", logger); + given(logger.isDebugEnabled()).willReturn(true); + final AtomicInteger attemptingReconnectCount = new AtomicInteger(); + willAnswer(i -> { + if (attemptingReconnectCount.getAndIncrement() == 0) { + adapter.connectionLost(new RuntimeException("while schedule running")); + } + i.callRealMethod(); + return null; + }).given(logger).debug("Attempting reconnect"); + ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); + taskScheduler.initialize(); + adapter.setTaskScheduler(taskScheduler); + adapter.start(); + adapter.connectionLost(new RuntimeException("initial")); + Thread.sleep(1000); + // the following assertion should be equalTo, but leq to protect against a slow CI server + assertThat(attemptingReconnectCount.get(), lessThanOrEqualTo(2)); + adapter.stop(); + taskScheduler.destroy(); + } + private MqttPahoMessageDrivenChannelAdapter buildAdapter(final MqttAsyncClient client, Boolean cleanSession, ConsumerStopAction action) throws MqttException, MqttSecurityException { DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory() { @@ -394,6 +430,7 @@ public class MqttAdapterTests { MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter("client", factory, "foo"); adapter.setApplicationEventPublisher(mock(ApplicationEventPublisher.class)); adapter.setOutputChannel(new NullChannel()); + adapter.setTaskScheduler(mock(TaskScheduler.class)); adapter.afterPropertiesSet(); return adapter; }