Prevent cursor from using more than 100% temp store

activemq-broker/src/main/java/org/apache/activemq/broker/region/BaseDestination.java

@@ -31,6 +31,7 @@
 import org.apache.activemq.broker.region.policy.SlowConsumerStrategy;
 import org.apache.activemq.command.ActiveMQDestination;
 import org.apache.activemq.command.ActiveMQTopic;
+import org.apache.activemq.command.ConsumerInfo;
 import org.apache.activemq.command.Message;
 import org.apache.activemq.command.MessageAck;
 import org.apache.activemq.command.MessageDispatchNotification;
@@ -578,6 +579,12 @@ public void messageDiscarded(ConnectionContext context, Subscription sub, Messag
         if (advisoryForDiscardingMessages) {
             broker.messageDiscarded(context, sub, messageReference);
         }
+        // We need to send to the DLQ because broker.messageDiscarded() will not do that because it's
+        // optionally enabled and off by default. This is different than expiration handling because
+        // broker.messageExpired() does send to the DLQ
+        final ConsumerInfo info = sub != null ? sub.getConsumerInfo() : null;
+        final String poisonCause = info != null ? "Subscription discard. ID:" + info.getConsumerId() : "Message discarded";
+        broker.sendToDeadLetterQueue(context, messageReference, sub, new Throwable(poisonCause));
     }
 
     /**

activemq-broker/src/main/java/org/apache/activemq/broker/region/TopicSubscription.java

@@ -82,7 +82,7 @@ public TopicSubscription(Broker broker,ConnectionContext context, ConsumerInfo i
         if (info.getDestination().isTemporary() || broker.getTempDataStore()==null ) {
             this.matched = new VMPendingMessageCursor(false);
         } else {
-            this.matched = new FilePendingMessageCursor(broker,matchedName,false);
+            this.matched = new FilePendingMessageCursor(broker,matchedName,false, this);
         }
 
         this.scheduler = broker.getScheduler();
@@ -759,19 +759,25 @@ private void discard(MessageReference message, boolean expired) {
                     destination.getDestinationStatistics().getNetworkDequeues().increment();
                 }
             }
-            Destination dest = (Destination) message.getRegionDestination();
+
+            final Destination dest = (Destination) message.getRegionDestination();
+            // This should not be null
             if (dest != null) {
                 //If discard is due to expiration then use the messageExpired() callback
                 if (expired) {
                     LOG.debug("{}, expiring message {}", this, message);
+                    // This callback will also send messages to the DLQ
                     dest.messageExpired(getContext(), this, message);
                 } else {
                     LOG.debug("{}, discarding message {}", this, message);
                     discarded.incrementAndGet();
+                    // This callback will also send messages to the DLQ
                     dest.messageDiscarded(getContext(), this, message);
                 }
+            } else {
+                LOG.debug("Message {} regionDestination unexpectedly unset during discard on {}",
+                        message.getMessageId(), info.getConsumerId());
             }
-            broker.getRoot().sendToDeadLetterQueue(getContext(), message, this, new Throwable("TopicSubDiscard. ID:" + info.getConsumerId()));
         } finally {
             discarding = false;
         }