Repository: moquette-io/moquette (java)

Issue:
immediate_buffer_flush should be in configuration file, defaulting to true.
### Expected behavior
Correct performance behavior, similar to Mosquitto or Aedes (less than 20ms for message to propagate from emitter to subscriber on localhost).
The parameter `immediate_buffer_flush` is not listed in the default conf file, and not mentioned in the documentation.
The default value does not seem appropriate for normal use cases.

### Actual behavior
Requests are buffered, resulting in delays of 1 second.

### Steps to reproduce
Run with default configuration.

### Minimal yet complete reproducer code (or URL to code) or complete log file

### Moquette MQTT version
0.16, 0.17-SNAPSHOT

### JVM version (e.g. `java -version`)
n/a

### OS version (e.g. `uname -a`)
n/a

Pull request description:
Switched boolean flag to immediate flushing to integer
## Release notes
Introduces the ability to specify the flush interval for IO write operations, changing the default, to an immediate flush.

## What does it do?
- Changed the value of configuration setting `immediate_buffer_flush` from boolean to a number (expressing milliseconds)
- Added 'buffer_flush_millis' setting which overrides the deprecated 'immediate_buffer_flush'
- Changed default behavior from "no flush on every write" to flush on each write.
- Used the value of the new `buffer_flush_millis` settings to initialize the auto flusher.

Fixes #718

Patch:
diff --git a/ChangeLog.txt b/ChangeLog.txt
index 2f12b37e..659eeffd 100644
--- a/ChangeLog.txt
+++ b/ChangeLog.txt
@@ -1,4 +1,5 @@
 Version 0.17-SNAPSHOT:
+   [feature] introduce new `buffer_flush_millis` deprecating the old `immediate_buffer_flush` and switch default behavior to flush on every write (#738).
    [refactory] purge of session state also on disconnect and reused logic (#715).
    [feature] add `moquette.session_loop.debug` property to enable session loop checking assignments (#714).
    [break] deprecate `persistent_store` to separate the enablement of persistence with `persistence_enabled` and the path `data_path` (#706).
diff --git a/broker/src/main/java/io/moquette/BrokerConstants.java b/broker/src/main/java/io/moquette/BrokerConstants.java
index 26929b32..d3e926d0 100644
--- a/broker/src/main/java/io/moquette/BrokerConstants.java
+++ b/broker/src/main/java/io/moquette/BrokerConstants.java
@@ -86,7 +86,20 @@ public final class BrokerConstants {
     public static final String NETTY_EPOLL_PROPERTY_NAME = "netty.epoll";
     public static final String NETTY_MAX_BYTES_PROPERTY_NAME = "netty.mqtt.message_size";
     public static final int DEFAULT_NETTY_MAX_BYTES_IN_MESSAGE = 8092;
+    /**
+     * @deprecated use the BUFFER_FLUSH_MS_PROPERTY_NAME
+     * */
+    @Deprecated
     public static final String IMMEDIATE_BUFFER_FLUSH_PROPERTY_NAME = "immediate_buffer_flush";
+    /**
+     * 0/immediate means immediate flush, like immediate_buffer_flush = true
+     * -1/full means no explicit flush, let Netty flush when write buffers are full, like immediate_buffer_flush = false
+     * a number of milliseconds to between flushes
+     * */
+    public static final String BUFFER_FLUSH_MS_PROPERTY_NAME = "buffer_flush_millis";
+    public static final int NO_BUFFER_FLUSH = -1;
+    public static final int IMMEDIATE_BUFFER_FLUSH = 0;
+
     public static final String METRICS_ENABLE_PROPERTY_NAME = "use_metrics";
     public static final String METRICS_LIBRATO_EMAIL_PROPERTY_NAME = "metrics.librato.email";
     public static final String METRICS_LIBRATO_TOKEN_PROPERTY_NAME = "metrics.librato.token";
diff --git a/broker/src/main/java/io/moquette/broker/BrokerConfiguration.java b/broker/src/main/java/io/moquette/broker/BrokerConfiguration.java
index ec312420..6d0c190b 100644
--- a/broker/src/main/java/io/moquette/broker/BrokerConfiguration.java
+++ b/broker/src/main/java/io/moquette/broker/BrokerConfiguration.java
@@ -18,26 +18,57 @@ package io.moquette.broker;
 import io.moquette.BrokerConstants;
 import io.moquette.broker.config.IConfig;
 
+import java.util.Locale;
+
 class BrokerConfiguration {
 
     private final boolean allowAnonymous;
     private final boolean allowZeroByteClientId;
     private final boolean reauthorizeSubscriptionsOnConnect;
-    private final boolean immediateBufferFlush;
+    private final int bufferFlushMillis;
 
     BrokerConfiguration(IConfig props) {
         allowAnonymous = props.boolProp(BrokerConstants.ALLOW_ANONYMOUS_PROPERTY_NAME, true);
         allowZeroByteClientId = props.boolProp(BrokerConstants.ALLOW_ZERO_BYTE_CLIENT_ID_PROPERTY_NAME, false);
         reauthorizeSubscriptionsOnConnect = props.boolProp(BrokerConstants.REAUTHORIZE_SUBSCRIPTIONS_ON_CONNECT, false);
-        immediateBufferFlush = props.boolProp(BrokerConstants.IMMEDIATE_BUFFER_FLUSH_PROPERTY_NAME, false);
+
+        // BUFFER_FLUSH_MS_PROPERTY_NAME has precedence over the deprecated IMMEDIATE_BUFFER_FLUSH_PROPERTY_NAME
+        final String bufferFlushMillisProp = props.getProperty(BrokerConstants.BUFFER_FLUSH_MS_PROPERTY_NAME);
+        if (bufferFlushMillisProp != null && !bufferFlushMillisProp.isEmpty()) {
+            switch (bufferFlushMillisProp.toLowerCase(Locale.ROOT)) {
+                case "immediate":
+                    bufferFlushMillis = BrokerConstants. IMMEDIATE_BUFFER_FLUSH;
+                    break;
+                case "full":
+                    bufferFlushMillis = BrokerConstants.NO_BUFFER_FLUSH;
+                    break;
+                default:
+                    final String errorMsg = String.format("Can't state value of %s property. Has to be 'immediate', " +
+                        "'full' or a number >= -1, found %s", BrokerConstants.BUFFER_FLUSH_MS_PROPERTY_NAME, bufferFlushMillisProp);
+                    try {
+                        bufferFlushMillis = Integer.parseInt(bufferFlushMillisProp);
+                        if (bufferFlushMillis < -1) {
+                            throw new IllegalArgumentException(errorMsg);
+                        }
+                    } catch (NumberFormatException ex) {
+                        throw new IllegalArgumentException(errorMsg);
+                    }
+            }
+        } else {
+            if (props.boolProp(BrokerConstants.IMMEDIATE_BUFFER_FLUSH_PROPERTY_NAME, true)) {
+                bufferFlushMillis = BrokerConstants.IMMEDIATE_BUFFER_FLUSH;
+            } else {
+                bufferFlushMillis = BrokerConstants.NO_BUFFER_FLUSH;
+            }
+        }
     }
 
     public BrokerConfiguration(boolean allowAnonymous, boolean allowZeroByteClientId,
-                               boolean reauthorizeSubscriptionsOnConnect, boolean immediateBufferFlush) {
+                               boolean reauthorizeSubscriptionsOnConnect, int bufferFlushMillis) {
         this.allowAnonymous = allowAnonymous;
         this.allowZeroByteClientId = allowZeroByteClientId;
         this.reauthorizeSubscriptionsOnConnect = reauthorizeSubscriptionsOnConnect;
-        this.immediateBufferFlush = immediateBufferFlush;
+        this.bufferFlushMillis = bufferFlushMillis;
     }
 
     public boolean isAllowAnonymous() {
@@ -52,7 +83,7 @@ class BrokerConfiguration {
         return reauthorizeSubscriptionsOnConnect;
     }
 
-    public boolean isImmediateBufferFlush() {
-        return immediateBufferFlush;
+    public int getBufferFlushMillis() {
+        return bufferFlushMillis;
     }
 }
diff --git a/broker/src/main/java/io/moquette/broker/MQTTConnection.java b/broker/src/main/java/io/moquette/broker/MQTTConnection.java
index 42eb0f38..892c980a 100644
--- a/broker/src/main/java/io/moquette/broker/MQTTConnection.java
+++ b/broker/src/main/java/io/moquette/broker/MQTTConnection.java
@@ -15,6 +15,7 @@
  */
 package io.moquette.broker;
 
+import io.moquette.BrokerConstants;
 import io.moquette.broker.subscriptions.Topic;
 import io.moquette.broker.security.IAuthenticator;
 import io.netty.buffer.ByteBuf;
@@ -535,7 +536,7 @@ final class MQTTConnection {
             }
 
             ChannelFuture channelFuture;
-            if (brokerConfig.isImmediateBufferFlush()) {
+            if (brokerConfig.getBufferFlushMillis() == BrokerConstants.IMMEDIATE_BUFFER_FLUSH) {
                 channelFuture = channel.writeAndFlush(retainedDup);
             } else {
                 channelFuture = channel.write(retainedDup);
diff --git a/broker/src/main/java/io/moquette/broker/NewNettyAcceptor.java b/broker/src/main/java/io/moquette/broker/NewNettyAcceptor.java
index 20293d66..25073829 100644
--- a/broker/src/main/java/io/moquette/broker/NewNettyAcceptor.java
+++ b/broker/src/main/java/io/moquette/broker/NewNettyAcceptor.java
@@ -137,7 +137,7 @@ class NewNettyAcceptor {
 
     private Class<? extends ServerSocketChannel> channelClass;
 
-    public void initialize(NewNettyMQTTHandler mqttHandler, IConfig props, ISslContextCreator sslCtxCreator) {
+    public void initialize(NewNettyMQTTHandler mqttHandler, IConfig props, ISslContextCreator sslCtxCreator, BrokerConfiguration brokerConfiguration) {
         LOG.debug("Initializing Netty acceptor");
 
         nettySoBacklog = props.intProp(BrokerConstants.NETTY_SO_BACKLOG_PROPERTY_NAME, 128);
@@ -178,16 +178,16 @@ class NewNettyAcceptor {
         } else {
             this.errorsCather = Optional.empty();
         }
-        initializePlainTCPTransport(mqttHandler, props);
-        initializeWebSocketTransport(mqttHandler, props);
+        initializePlainTCPTransport(mqttHandler, props, brokerConfiguration);
+        initializeWebSocketTransport(mqttHandler, props, brokerConfiguration);
         if (securityPortsConfigured(props)) {
             SslContext sslContext = sslCtxCreator.initSSLContext();
             if (sslContext == null) {
                 LOG.error("Can't initialize SSLHandler layer! Exiting, check your configuration of jks");
                 return;
             }
-            initializeSSLTCPTransport(mqttHandler, props, sslContext);
-            initializeWSSTransport(mqttHandler, props, sslContext);
+            initializeSSLTCPTransport(mqttHandler, props, sslContext, brokerConfiguration);
+            initializeWSSTransport(mqttHandler, props, sslContext, brokerConfiguration);
         }
     }
 
@@ -239,7 +239,7 @@ class NewNettyAcceptor {
         return ports.computeIfAbsent(SSL_MQTT_PROTO, i -> 0);
     }
 
-    private void initializePlainTCPTransport(NewNettyMQTTHandler handler, IConfig props) {
+    private void initializePlainTCPTransport(NewNettyMQTTHandler handler, IConfig props, BrokerConfiguration brokerConfiguration) {
         LOG.debug("Configuring TCP MQTT transport");
         final MoquetteIdleTimeoutHandler timeoutHandler = new MoquetteIdleTimeoutHandler();
         String host = props.getProperty(BrokerConstants.HOST_PROPERTY_NAME);
@@ -250,18 +250,19 @@ class NewNettyAcceptor {
             return;
         }
         int port = Integer.parseInt(tcpPortProp);
+        final int writeFlushMillis = brokerConfiguration.getBufferFlushMillis();
         initFactory(host, port, PLAIN_MQTT_PROTO, new PipelineInitializer() {
 
             @Override
             void init(SocketChannel channel) {
                 ChannelPipeline pipeline = channel.pipeline();
-                configureMQTTPipeline(pipeline, timeoutHandler, handler);
+                configureMQTTPipeline(pipeline, timeoutHandler, handler, writeFlushMillis);
             }
         });
     }
 
     private void configureMQTTPipeline(ChannelPipeline pipeline, MoquetteIdleTimeoutHandler timeoutHandler,
-                                       NewNettyMQTTHandler handler) {
+                                       NewNettyMQTTHandler handler, int writeFlushMillis) {
         pipeline.addFirst("idleStateHandler", new IdleStateHandler(nettyChannelTimeoutSeconds, 0, 0));
         pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler);
         // pipeline.addLast("logger", new LoggingHandler("Netty", LogLevel.ERROR));
@@ -269,7 +270,9 @@ class NewNettyAcceptor {
             pipeline.addLast("bugsnagCatcher", errorsCather.get());
         }
         pipeline.addFirst("bytemetrics", new BytesMetricsHandler(bytesMetricsCollector));
-        pipeline.addLast("autoflush", new AutoFlushHandler(1, TimeUnit.SECONDS));
+        if (writeFlushMillis > IMMEDIATE_BUFFER_FLUSH) {
+            pipeline.addLast("autoflush", new AutoFlushHandler(writeFlushMillis, TimeUnit.MILLISECONDS));
+        }
         pipeline.addLast("decoder", new MqttDecoder(maxBytesInMessage));
         pipeline.addLast("encoder", MqttEncoder.INSTANCE);
         pipeline.addLast("metrics", new MessageMetricsHandler(metricsCollector));
@@ -280,7 +283,7 @@ class NewNettyAcceptor {
         pipeline.addLast("handler", handler);
     }
 
-    private void initializeWebSocketTransport(final NewNettyMQTTHandler handler, IConfig props) {
+    private void initializeWebSocketTransport(final NewNettyMQTTHandler handler, IConfig props, BrokerConfiguration brokerConfiguration) {
         LOG.debug("Configuring Websocket MQTT transport");
         String webSocketPortProp = props.getProperty(WEB_SOCKET_PORT_PROPERTY_NAME, DISABLED_PORT_BIND);
         if (DISABLED_PORT_BIND.equals(webSocketPortProp)) {
@@ -296,6 +299,7 @@ class NewNettyAcceptor {
         String host = props.getProperty(BrokerConstants.HOST_PROPERTY_NAME);
         String path = props.getProperty(BrokerConstants.WEB_SOCKET_PATH_PROPERTY_NAME, BrokerConstants.WEBSOCKET_PATH);
         int maxFrameSize = props.intProp(BrokerConstants.WEB_SOCKET_MAX_FRAME_SIZE_PROPERTY_NAME, 65536);
+        final int writeFlushMillis = brokerConfiguration.getBufferFlushMillis();
         initFactory(host, port, "Websocket MQTT", new PipelineInitializer() {
 
             @Override
@@ -307,12 +311,12 @@ class NewNettyAcceptor {
                         new WebSocketServerProtocolHandler(path, MQTT_SUBPROTOCOL_CSV_LIST, false, maxFrameSize));
                 pipeline.addLast("ws2bytebufDecoder", new WebSocketFrameToByteBufDecoder());
                 pipeline.addLast("bytebuf2wsEncoder", new ByteBufToWebSocketFrameEncoder());
-                configureMQTTPipeline(pipeline, timeoutHandler, handler);
+                configureMQTTPipeline(pipeline, timeoutHandler, handler, writeFlushMillis);
             }
         });
     }
 
-    private void initializeSSLTCPTransport(NewNettyMQTTHandler handler, IConfig props, SslContext sslContext) {
+    private void initializeSSLTCPTransport(NewNettyMQTTHandler handler, IConfig props, SslContext sslContext, BrokerConfiguration brokerConfiguration) {
         LOG.debug("Configuring SSL MQTT transport");
         String sslPortProp = props.getProperty(SSL_PORT_PROPERTY_NAME, DISABLED_PORT_BIND);
         if (DISABLED_PORT_BIND.equals(sslPortProp)) {
@@ -329,18 +333,19 @@ class NewNettyAcceptor {
         String host = props.getProperty(BrokerConstants.HOST_PROPERTY_NAME);
         String sNeedsClientAuth = props.getProperty(BrokerConstants.NEED_CLIENT_AUTH, "false");
         final boolean needsClientAuth = Boolean.valueOf(sNeedsClientAuth);
+        final int writeFlushMillis = brokerConfiguration.getBufferFlushMillis();
         initFactory(host, sslPort, SSL_MQTT_PROTO, new PipelineInitializer() {
 
             @Override
             void init(SocketChannel channel) throws Exception {
                 ChannelPipeline pipeline = channel.pipeline();
                 pipeline.addLast("ssl", createSslHandler(channel, sslContext, needsClientAuth));
-                configureMQTTPipeline(pipeline, timeoutHandler, handler);
+                configureMQTTPipeline(pipeline, timeoutHandler, handler, writeFlushMillis);
             }
         });
     }
 
-    private void initializeWSSTransport(NewNettyMQTTHandler handler, IConfig props, SslContext sslContext) {
+    private void initializeWSSTransport(NewNettyMQTTHandler handler, IConfig props, SslContext sslContext, BrokerConfiguration brokerConfiguration) {
         LOG.debug("Configuring secure websocket MQTT transport");
         String sslPortProp = props.getProperty(WSS_PORT_PROPERTY_NAME, DISABLED_PORT_BIND);
         if (DISABLED_PORT_BIND.equals(sslPortProp)) {
@@ -356,6 +361,7 @@ class NewNettyAcceptor {
         int maxFrameSize = props.intProp(BrokerConstants.WEB_SOCKET_MAX_FRAME_SIZE_PROPERTY_NAME, 65536);
         String sNeedsClientAuth = props.getProperty(BrokerConstants.NEED_CLIENT_AUTH, "false");
         final boolean needsClientAuth = Boolean.valueOf(sNeedsClientAuth);
+        final int writeFlushMillis = brokerConfiguration.getBufferFlushMillis();
         initFactory(host, sslPort, "Secure websocket", new PipelineInitializer() {
 
             @Override
@@ -370,7 +376,7 @@ class NewNettyAcceptor {
                 pipeline.addLast("ws2bytebufDecoder", new WebSocketFrameToByteBufDecoder());
                 pipeline.addLast("bytebuf2wsEncoder", new ByteBufToWebSocketFrameEncoder());
 
-                configureMQTTPipeline(pipeline, timeoutHandler, handler);
+                configureMQTTPipeline(pipeline, timeoutHandler, handler, writeFlushMillis);
             }
         });
     }
diff --git a/broker/src/main/java/io/moquette/broker/Server.java b/broker/src/main/java/io/moquette/broker/Server.java
index 2b9e8dcb..fe933e29 100644
--- a/broker/src/main/java/io/moquette/broker/Server.java
+++ b/broker/src/main/java/io/moquette/broker/Server.java
@@ -249,7 +249,7 @@ public class Server {
 
         final NewNettyMQTTHandler mqttHandler = new NewNettyMQTTHandler(connectionFactory);
         acceptor = new NewNettyAcceptor();
-        acceptor.initialize(mqttHandler, config, sslCtxCreator);
+        acceptor.initialize(mqttHandler, config, sslCtxCreator, brokerConfig);
 
         final long startTime = System.currentTimeMillis() - start;
         LOG.info("Moquette integration has been started successfully in {} ms", startTime);
diff --git a/distribution/src/main/resources/moquette.conf b/distribution/src/main/resources/moquette.conf
index 660f3107..40116ce0 100644
--- a/distribution/src/main/resources/moquette.conf
+++ b/distribution/src/main/resources/moquette.conf
@@ -186,3 +186,14 @@ password_file config/password_file.conf
 # default: true
 #*********************************************************************
 # telemetry_enabled true
+
+#*********************************************************************
+# Flush interval between writes
+#
+# buffer_flush_millis:
+#       `immediate` or `full` or number. `immediate` forces the flush on
+#       every socket write while `full` let the underlying system to flush
+#       when full. If its defined a number it's used a milliseconds between flushes.
+# default: immediate
+#*********************************************************************
+# buffer_flush_millis immediate


Test patch:
diff --git a/broker/src/test/java/io/moquette/broker/BrokerConfigurationTest.java b/broker/src/test/java/io/moquette/broker/BrokerConfigurationTest.java
index b4e273dc..b8d8ec8c 100644
--- a/broker/src/test/java/io/moquette/broker/BrokerConfigurationTest.java
+++ b/broker/src/test/java/io/moquette/broker/BrokerConfigurationTest.java
@@ -21,6 +21,7 @@ import org.junit.jupiter.api.Test;
 
 import java.util.Properties;
 
+import static io.moquette.BrokerConstants.IMMEDIATE_BUFFER_FLUSH;
 import static org.junit.jupiter.api.Assertions.*;
 
 public class BrokerConfigurationTest {
@@ -32,7 +33,7 @@ public class BrokerConfigurationTest {
         assertTrue(brokerConfiguration.isAllowAnonymous());
         assertFalse(brokerConfiguration.isAllowZeroByteClientId());
         assertFalse(brokerConfiguration.isReauthorizeSubscriptionsOnConnect());
-        assertFalse(brokerConfiguration.isImmediateBufferFlush());
+        assertEquals(IMMEDIATE_BUFFER_FLUSH, brokerConfiguration.getBufferFlushMillis(), "Immediate flush by default");
     }
 
     @Test
@@ -44,7 +45,7 @@ public class BrokerConfigurationTest {
         assertFalse(brokerConfiguration.isAllowAnonymous());
         assertFalse(brokerConfiguration.isAllowZeroByteClientId());
         assertFalse(brokerConfiguration.isReauthorizeSubscriptionsOnConnect());
-        assertFalse(brokerConfiguration.isImmediateBufferFlush());
+        assertEquals(IMMEDIATE_BUFFER_FLUSH, brokerConfiguration.getBufferFlushMillis(), "Immediate flush by default");
     }
 
     @Test
@@ -56,7 +57,7 @@ public class BrokerConfigurationTest {
         assertTrue(brokerConfiguration.isAllowAnonymous());
         assertTrue(brokerConfiguration.isAllowZeroByteClientId());
         assertFalse(brokerConfiguration.isReauthorizeSubscriptionsOnConnect());
-        assertFalse(brokerConfiguration.isImmediateBufferFlush());
+        assertEquals(IMMEDIATE_BUFFER_FLUSH, brokerConfiguration.getBufferFlushMillis(), "Immediate flush by default");
     }
 
     @Test
@@ -68,7 +69,7 @@ public class BrokerConfigurationTest {
         assertTrue(brokerConfiguration.isAllowAnonymous());
         assertFalse(brokerConfiguration.isAllowZeroByteClientId());
         assertTrue(brokerConfiguration.isReauthorizeSubscriptionsOnConnect());
-        assertFalse(brokerConfiguration.isImmediateBufferFlush());
+        assertEquals(IMMEDIATE_BUFFER_FLUSH, brokerConfiguration.getBufferFlushMillis(), "Immediate flush by default");
     }
 
     @Test
@@ -80,6 +81,6 @@ public class BrokerConfigurationTest {
         assertTrue(brokerConfiguration.isAllowAnonymous());
         assertFalse(brokerConfiguration.isAllowZeroByteClientId());
         assertFalse(brokerConfiguration.isReauthorizeSubscriptionsOnConnect());
-        assertTrue(brokerConfiguration.isImmediateBufferFlush());
+        assertEquals(IMMEDIATE_BUFFER_FLUSH, brokerConfiguration.getBufferFlushMillis(), "No immediate flush by default");
     }
 }
diff --git a/broker/src/test/java/io/moquette/broker/MQTTConnectionConnectTest.java b/broker/src/test/java/io/moquette/broker/MQTTConnectionConnectTest.java
index 8f37c8cc..3dc15631 100644
--- a/broker/src/test/java/io/moquette/broker/MQTTConnectionConnectTest.java
+++ b/broker/src/test/java/io/moquette/broker/MQTTConnectionConnectTest.java
@@ -32,6 +32,7 @@ import org.junit.jupiter.api.Test;
 import java.util.concurrent.ExecutionException;
 
 import static io.moquette.broker.MQTTConnectionPublishTest.memorySessionsRepository;
+import static io.moquette.BrokerConstants.NO_BUFFER_FLUSH;
 import static io.moquette.broker.NettyChannelAssertions.assertEqualsConnAck;
 import static io.netty.handler.codec.mqtt.MqttConnectReturnCode.*;
 import static java.util.Collections.singleton;
@@ -53,7 +54,7 @@ public class MQTTConnectionConnectTest {
     private EmbeddedChannel channel;
     private SessionRegistry sessionRegistry;
     private MqttMessageBuilders.ConnectBuilder connMsg;
-    private static final BrokerConfiguration CONFIG = new BrokerConfiguration(true, true, false, false);
+    private static final BrokerConfiguration CONFIG = new BrokerConfiguration(true, true, false, NO_BUFFER_FLUSH);
     private IAuthenticator mockAuthenticator;
     private PostOffice postOffice;
     private MemoryQueueRepository queueRepository;
@@ -209,7 +210,7 @@ public class MQTTConnectionConnectTest {
     @Test
     public void prohibitAnonymousClient() {
         MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID).build();
-        BrokerConfiguration config = new BrokerConfiguration(false, true, false, false);
+        BrokerConfiguration config = new BrokerConfiguration(false, true, false, NO_BUFFER_FLUSH);
 
         sut = createMQTTConnection(config);
         channel = (EmbeddedChannel) sut.channel;
@@ -227,7 +228,7 @@ public class MQTTConnectionConnectTest {
         MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID)
             .username(TEST_USER + "_fake")
             .build();
-        BrokerConfiguration config = new BrokerConfiguration(false, true, false, false);
+        BrokerConfiguration config = new BrokerConfiguration(false, true, false, NO_BUFFER_FLUSH);
 
         createMQTTConnection(config);
 
@@ -241,7 +242,7 @@ public class MQTTConnectionConnectTest {
 
     @Test
     public void testZeroByteClientIdNotAllowed() {
-        BrokerConfiguration config = new BrokerConfiguration(false, false, false, false);
+        BrokerConfiguration config = new BrokerConfiguration(false, false, false, NO_BUFFER_FLUSH);
 
         sut = createMQTTConnection(config);
         channel = (EmbeddedChannel) sut.channel;
@@ -292,7 +293,7 @@ public class MQTTConnectionConnectTest {
         EmbeddedChannel evilChannel = new EmbeddedChannel();
 
         // Exercise
-        BrokerConfiguration config = new BrokerConfiguration(true, true, false, false);
+        BrokerConfiguration config = new BrokerConfiguration(true, true, false, NO_BUFFER_FLUSH);
         final MQTTConnection evilConnection = createMQTTConnection(config, evilChannel, postOffice);
         evilConnection.processConnect(evilClientConnMsg);
 
diff --git a/broker/src/test/java/io/moquette/broker/MQTTConnectionPublishTest.java b/broker/src/test/java/io/moquette/broker/MQTTConnectionPublishTest.java
index 593f96db..4fc61648 100644
--- a/broker/src/test/java/io/moquette/broker/MQTTConnectionPublishTest.java
+++ b/broker/src/test/java/io/moquette/broker/MQTTConnectionPublishTest.java
@@ -35,6 +35,7 @@ import org.junit.jupiter.api.Test;
 
 import java.util.concurrent.ExecutionException;
 
+import static io.moquette.BrokerConstants.NO_BUFFER_FLUSH;
 import static java.nio.charset.StandardCharsets.UTF_8;
 import static java.util.Collections.singleton;
 import static java.util.Collections.singletonMap;
@@ -56,7 +57,7 @@ public class MQTTConnectionPublishTest {
     public void setUp() {
         connMsg = MqttMessageBuilders.connect().protocolVersion(MqttVersion.MQTT_3_1).cleanSession(true);
 
-        BrokerConfiguration config = new BrokerConfiguration(true, true, false, false);
+        BrokerConfiguration config = new BrokerConfiguration(true, true, false, NO_BUFFER_FLUSH);
 
         createMQTTConnection(config);
     }
diff --git a/broker/src/test/java/io/moquette/broker/PostOfficeInternalPublishTest.java b/broker/src/test/java/io/moquette/broker/PostOfficeInternalPublishTest.java
index 181ada1d..83ab4210 100644
--- a/broker/src/test/java/io/moquette/broker/PostOfficeInternalPublishTest.java
+++ b/broker/src/test/java/io/moquette/broker/PostOfficeInternalPublishTest.java
@@ -33,6 +33,7 @@ import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 
 import static io.moquette.broker.MQTTConnectionPublishTest.memorySessionsRepository;
+import static io.moquette.BrokerConstants.NO_BUFFER_FLUSH;
 import static io.moquette.broker.PostOfficeUnsubscribeTest.CONFIG;
 import static io.netty.handler.codec.mqtt.MqttQoS.*;
 import static java.nio.charset.StandardCharsets.UTF_8;
@@ -55,7 +56,7 @@ public class PostOfficeInternalPublishTest {
     private SessionRegistry sessionRegistry;
     private MockAuthenticator mockAuthenticator;
     private static final BrokerConfiguration ALLOW_ANONYMOUS_AND_ZERO_BYTES_CLID =
-        new BrokerConfiguration(true, true, false, false);
+        new BrokerConfiguration(true, true, false, NO_BUFFER_FLUSH);
     private MemoryRetainedRepository retainedRepository;
     private MemoryQueueRepository queueRepository;
 
diff --git a/broker/src/test/java/io/moquette/broker/PostOfficePublishTest.java b/broker/src/test/java/io/moquette/broker/PostOfficePublishTest.java
index a88e6a72..a085b6d9 100644
--- a/broker/src/test/java/io/moquette/broker/PostOfficePublishTest.java
+++ b/broker/src/test/java/io/moquette/broker/PostOfficePublishTest.java
@@ -39,6 +39,7 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
 import static io.moquette.broker.MQTTConnectionPublishTest.memorySessionsRepository;
+import static io.moquette.BrokerConstants.NO_BUFFER_FLUSH;
 import static io.moquette.broker.PostOfficeUnsubscribeTest.CONFIG;
 import static io.netty.handler.codec.mqtt.MqttQoS.AT_LEAST_ONCE;
 import static io.netty.handler.codec.mqtt.MqttQoS.AT_MOST_ONCE;
@@ -67,7 +68,7 @@ public class PostOfficePublishTest {
     private SessionRegistry sessionRegistry;
     private MockAuthenticator mockAuthenticator;
     static final BrokerConfiguration ALLOW_ANONYMOUS_AND_ZERO_BYTES_CLID =
-        new BrokerConfiguration(true, true, false, false);
+        new BrokerConfiguration(true, true, false, NO_BUFFER_FLUSH);
     private MemoryRetainedRepository retainedRepository;
     private MemoryQueueRepository queueRepository;
 
diff --git a/broker/src/test/java/io/moquette/broker/PostOfficeSubscribeTest.java b/broker/src/test/java/io/moquette/broker/PostOfficeSubscribeTest.java
index 2b841716..930b86ce 100644
--- a/broker/src/test/java/io/moquette/broker/PostOfficeSubscribeTest.java
+++ b/broker/src/test/java/io/moquette/broker/PostOfficeSubscribeTest.java
@@ -39,6 +39,7 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
 import static io.moquette.broker.MQTTConnectionPublishTest.memorySessionsRepository;
+import static io.moquette.BrokerConstants.NO_BUFFER_FLUSH;
 import static io.moquette.broker.PostOfficePublishTest.ALLOW_ANONYMOUS_AND_ZERO_BYTES_CLID;
 import static io.moquette.broker.PostOfficePublishTest.SUBSCRIBER_ID;
 import static io.netty.handler.codec.mqtt.MqttConnectReturnCode.CONNECTION_ACCEPTED;
@@ -68,7 +69,7 @@ public class PostOfficeSubscribeTest {
     private MqttConnectMessage connectMessage;
     private IAuthenticator mockAuthenticator;
     private SessionRegistry sessionRegistry;
-    public static final BrokerConfiguration CONFIG = new BrokerConfiguration(true, true, false, false);
+    public static final BrokerConfiguration CONFIG = new BrokerConfiguration(true, true, false, NO_BUFFER_FLUSH);
     private MemoryQueueRepository queueRepository;
 
     @BeforeEach
diff --git a/broker/src/test/java/io/moquette/broker/PostOfficeUnsubscribeTest.java b/broker/src/test/java/io/moquette/broker/PostOfficeUnsubscribeTest.java
index 8a7d68a9..f1494ea3 100644
--- a/broker/src/test/java/io/moquette/broker/PostOfficeUnsubscribeTest.java
+++ b/broker/src/test/java/io/moquette/broker/PostOfficeUnsubscribeTest.java
@@ -38,6 +38,7 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
 import static io.moquette.broker.MQTTConnectionPublishTest.memorySessionsRepository;
+import static io.moquette.BrokerConstants.NO_BUFFER_FLUSH;
 import static io.moquette.broker.PostOfficePublishTest.PUBLISHER_ID;
 import static io.netty.handler.codec.mqtt.MqttQoS.*;
 import static java.util.Collections.*;
@@ -59,7 +60,7 @@ public class PostOfficeUnsubscribeTest {
     private MqttConnectMessage connectMessage;
     private IAuthenticator mockAuthenticator;
     private SessionRegistry sessionRegistry;
-    public static final BrokerConfiguration CONFIG = new BrokerConfiguration(true, true, false, false);
+    public static final BrokerConfiguration CONFIG = new BrokerConfiguration(true, true, false, NO_BUFFER_FLUSH);
     private MemoryQueueRepository queueRepository;
 
     @BeforeEach
diff --git a/broker/src/test/java/io/moquette/broker/SessionRegistryTest.java b/broker/src/test/java/io/moquette/broker/SessionRegistryTest.java
index 36bda39b..c60d09f1 100644
--- a/broker/src/test/java/io/moquette/broker/SessionRegistryTest.java
+++ b/broker/src/test/java/io/moquette/broker/SessionRegistryTest.java
@@ -41,6 +41,7 @@ import java.nio.charset.StandardCharsets;
 import java.util.concurrent.ExecutionException;
 
 import static io.moquette.broker.MQTTConnectionPublishTest.memorySessionsRepository;
+import static io.moquette.BrokerConstants.NO_BUFFER_FLUSH;
 import static io.moquette.broker.NettyChannelAssertions.assertEqualsConnAck;
 import static io.netty.handler.codec.mqtt.MqttConnectReturnCode.CONNECTION_ACCEPTED;
 import static java.util.Collections.singleton;
@@ -60,7 +61,7 @@ public class SessionRegistryTest {
     private SessionRegistry sut;
     private MqttMessageBuilders.ConnectBuilder connMsg;
     private static final BrokerConfiguration ALLOW_ANONYMOUS_AND_ZEROBYTE_CLIENT_ID =
-        new BrokerConfiguration(true, true, false, false);
+        new BrokerConfiguration(true, true, false, NO_BUFFER_FLUSH);
     private MemoryQueueRepository queueRepository;
 
     @BeforeEach
diff --git a/broker/src/test/java/io/moquette/broker/SessionTest.java b/broker/src/test/java/io/moquette/broker/SessionTest.java
index 8d3b746f..00fb2b63 100644
--- a/broker/src/test/java/io/moquette/broker/SessionTest.java
+++ b/broker/src/test/java/io/moquette/broker/SessionTest.java
@@ -18,6 +18,8 @@ import org.assertj.core.api.Assertions;
 import static io.moquette.broker.Session.INFINITE_EXPIRY;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertTrue;
+import static io.moquette.BrokerConstants.NO_BUFFER_FLUSH;
+import static org.junit.jupiter.api.Assertions.*;
 
 public class SessionTest {
 
@@ -119,7 +121,7 @@ public class SessionTest {
     }
 
     private void createConnection(Session client) {
-        BrokerConfiguration brokerConfiguration = new BrokerConfiguration(true, false, false, false);
+        BrokerConfiguration brokerConfiguration = new BrokerConfiguration(true, false, false, NO_BUFFER_FLUSH);
         MQTTConnection mqttConnection = new MQTTConnection(testChannel, brokerConfiguration, null, null, null);
         client.markConnecting();
         client.bind(mqttConnection);