diff --git a/broker-core/src/main/java/org/apache/qpid/server/model/Broker.java b/broker-core/src/main/java/org/apache/qpid/server/model/Broker.java
index 86a8d09..533d6e2 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/model/Broker.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/model/Broker.java
@@ -371,6 +371,9 @@
 
     ScheduledFuture<?> scheduleTask(long delay, final TimeUnit unit, Runnable task);
 
+    @DerivedAttribute(description = "Threshold direct memory size (in bytes) at which the Broker will start flowing incoming messages to disk.")
+    long getFlowToDiskThreshold();
+
     @DerivedAttribute(description = "Threshold direct memory size (in bytes) at which the Broker will start considering to compact sparse buffers. Set to -1 to disable. See also " + MEMORY_OCCUPANCY_THRESHOLD)
     long getCompactMemoryThreshold();
 
diff --git a/broker-core/src/main/java/org/apache/qpid/server/model/BrokerImpl.java b/broker-core/src/main/java/org/apache/qpid/server/model/BrokerImpl.java
index 0a28e2b..c5c4aea 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/model/BrokerImpl.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/model/BrokerImpl.java
@@ -152,6 +152,7 @@
     private long _compactMemoryInterval;
     private double _memoryOccupancyThreshold;
     private long _memoryCompactionIncrement;
+    private long _flowToDiskThreshold;
 
     @ManagedObjectFactoryConstructor
     public BrokerImpl(Map<String, Object> attributes,
@@ -518,8 +519,7 @@
     @Override
     public synchronized void assignTargetSizes()
     {
-        long totalTarget = getContextValue(Long.class, BROKER_FLOW_TO_DISK_THRESHOLD);
-        LOGGER.debug("Assigning target sizes based on total target {}", totalTarget);
+        LOGGER.debug("Assigning target sizes based on total target {}", _flowToDiskThreshold);
         long totalSize = 0l;
         Collection<VirtualHostNode<?>> vhns = getVirtualHostNodes();
         Map<QueueManagingVirtualHost<?>, Long> vhs = new HashMap<>();
@@ -535,20 +535,20 @@
             }
         }
 
-        if (totalSize > totalTarget && !_totalMessageSizeExceedThresholdReported)
+        if (totalSize > _flowToDiskThreshold && !_totalMessageSizeExceedThresholdReported)
         {
-            _eventLogger.message(BrokerMessages.FLOW_TO_DISK_ACTIVE(totalSize / 1024, totalTarget / 1024));
+            _eventLogger.message(BrokerMessages.FLOW_TO_DISK_ACTIVE(totalSize / 1024, _flowToDiskThreshold / 1024));
             _totalMessageSizeExceedThresholdReported = true;
             _totalMessageSizeWithinThresholdReported = false;
         }
-        else if (totalSize <= totalTarget && !_totalMessageSizeWithinThresholdReported)
+        else if (totalSize <= _flowToDiskThreshold && !_totalMessageSizeWithinThresholdReported)
         {
-            _eventLogger.message(BrokerMessages.FLOW_TO_DISK_INACTIVE(totalSize / 1024, totalTarget / 1024));
+            _eventLogger.message(BrokerMessages.FLOW_TO_DISK_INACTIVE(totalSize / 1024, _flowToDiskThreshold / 1024));
             _totalMessageSizeWithinThresholdReported = true;
             _totalMessageSizeExceedThresholdReported = false;
         }
 
-        final long proportionalShare = (long) ((double) totalTarget / (double) vhs.size());
+        final long proportionalShare = (long) ((double) _flowToDiskThreshold / (double) vhs.size());
         for (Map.Entry<QueueManagingVirtualHost<?>, Long> entry : vhs.entrySet())
         {
             long virtualHostTotalQueueSize = entry.getValue();
@@ -559,7 +559,7 @@
             }
             else
             {
-                long queueSizeBasedShare = (totalTarget * virtualHostTotalQueueSize) / (2 * totalSize);
+                long queueSizeBasedShare = (_flowToDiskThreshold * virtualHostTotalQueueSize) / (2 * totalSize);
                 size = queueSizeBasedShare + (proportionalShare / 2);
             }
 
@@ -594,6 +594,7 @@
         long heapMemory = Runtime.getRuntime().maxMemory();
         getEventLogger().message(BrokerMessages.MAX_MEMORY(heapMemory, directMemory));
 
+        _flowToDiskThreshold = getContextValue(Long.class, BROKER_FLOW_TO_DISK_THRESHOLD);
         _compactMemoryThreshold = getContextValue(Long.class, Broker.COMPACT_MEMORY_THRESHOLD);
         _compactMemoryInterval = getContextValue(Long.class, Broker.COMPACT_MEMORY_INTERVAL);
         _memoryOccupancyThreshold = getContextValue(Double.class, Broker.MEMORY_OCCUPANCY_THRESHOLD);
@@ -854,6 +855,12 @@
     }
 
     @Override
+    public long getFlowToDiskThreshold()
+    {
+        return _flowToDiskThreshold;
+    }
+
+    @Override
     public long getNumberOfActivePooledBuffers()
     {
         return QpidByteBuffer.getNumberOfActivePooledBuffers();
diff --git a/broker-core/src/main/java/org/apache/qpid/server/queue/AbstractQueue.java b/broker-core/src/main/java/org/apache/qpid/server/queue/AbstractQueue.java
index 92a994d..2a46c82 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/queue/AbstractQueue.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/queue/AbstractQueue.java
@@ -65,6 +65,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.apache.qpid.server.bytebuffer.QpidByteBuffer;
 import org.apache.qpid.server.bytebuffer.QpidByteBufferInputStream;
 import org.apache.qpid.server.configuration.IllegalConfigurationException;
 import org.apache.qpid.server.configuration.updater.Task;
@@ -265,6 +266,7 @@
     private AdvanceConsumersTask _queueHouseKeepingTask;
     private volatile int _bindingCount;
     private volatile OverflowPolicyHandler _overflowPolicyHandler;
+    private long _flowToDiskThreshold;
 
     private interface HoldMethod
     {
@@ -469,6 +471,7 @@
         }
 
         _mimeTypeToFileExtension = getContextValue(Map.class, MAP_OF_STRING_STRING, MIME_TYPE_TO_FILE_EXTENSION);
+        _flowToDiskThreshold = getAncestor(Broker.class).getFlowToDiskThreshold();
 
         if(_defaultFilters != null)
         {
@@ -3332,7 +3335,9 @@
 
         void flowToDiskIfNecessary(StoredMessage<?> storedMessage, long estimatedQueueSize, final long targetQueueSize)
         {
-            if ((estimatedQueueSize > targetQueueSize) && storedMessage.isInMemory())
+            if ((estimatedQueueSize > targetQueueSize
+                 || QpidByteBuffer.getAllocatedDirectMemorySize() > _flowToDiskThreshold)
+                && storedMessage.isInMemory())
             {
                 storedMessage.flowToDisk();
             }
@@ -3348,7 +3353,8 @@
 
         void reportFlowToDiskStatusIfNecessary(final long estimatedQueueSize, final long targetQueueSize)
         {
-            if (estimatedQueueSize > targetQueueSize)
+            if (estimatedQueueSize > targetQueueSize
+                || QpidByteBuffer.getAllocatedDirectMemorySize() > _flowToDiskThreshold)
             {
                 reportFlowToDiskActiveIfNecessary(estimatedQueueSize, targetQueueSize);
             }