Conversation
… advisory handling - serviceRemoteConsumerAdvisory() used an exact class check (data.getClass() == ConsumerInfo.class) which silently drops any ConsumerInfo subclass. Change to instanceof so subclasses are accepted by the advisory processing path. - DemandForwardingBridgeSupport for subscriptions with same name
When multiple consumers share a durable subscription and ack messages out of order, a blind overwrite of LastAck can move the ack position backward. On restart, the recovery cursor replays already-acked messages. Check the update with a forward-only check so LastAck only advances. Test demonstrates recovery cost when lastAckedSequence regresses due to blind overwrite during out-of-order acks. Two durable subscriptions share a topic; the fast subscription acks 499 of 500 messages in reverse order. Without the forward-only guard in MessageDatabase, lastAck regresses to sequence 1 and recovery scans all 499 entries.
…upport Introduces protocol version 13, extending v12 with the fields needed to carry JMS 3.1 shared subscription state across the wire and through KahaDB recovery: ConsumerInfo + shared, durable (via SharedConsumerInfo) SubscriptionInfo + shared (via SharedSubscriptionInfo) ExceptionResponse + errorCode Message + deliveryTime The v13 MarshallerFactory clones the v12 marshaller map and replaces only the ten marshallers affected. OpenWireFormat already resolves the factory reflectively by version, so no version switch is required. Because OpenWire marshals a throwable as type and message only, the JMSException errorCode does not survive the round trip. The separate errorCode field carries it, and ExceptionResponse.getException() reapplies it to the reconstructed exception. Nothing here activates at the default store version: v13 is selected only when a broker sets storeOpenWireVersion to 13. Tests assert that gate directly -- deliveryTime, errorCode and the shared flags all fail to cross a v12 wire, and v12 continues to yield plain ConsumerInfo and SubscriptionInfo instances. ActiveMQErrorCode and SharedSubscriptionKey have no callers yet; they are scaffolding for the shared subscription broker work that follows. Adds 89 tests: round trips for all ten replaced marshallers in tight and loose encodings, the errorCode reconstruction branches, and null-clientId handling in SharedSubscriptionKey.
- SharedTopicBrokerService (server-side entry point) - SharedTopicConnectionFactory (client-side entry point) - SharedJMSContext wires the JMS 3.1 simplified-API shared consumer methods to SharedTopicSession. ActiveMQConnectionFactory and ActiveMQContext gain protected newContext/newChildContext hooks so the shared factory and context override construction only, not the createContext methods themselves. - A consumer may only join an active shared (or shared durable) subscription that has the same topic and selector. Joining under an in-use name with a different topic is rejected with ActiveMQErrorCode.TOPIC_MISMATCH, as Jakarta Messaging requires a JMSException/JMSRuntimeException there instead of silently joining.
Run the Jakarta Messaging TCK against SharedTopicBrokerService and SharedTopicConnectionFactory. The factory sets an explicit prefetch policy (durableTopicPrefetch must be >= 1: shared durable consumers become pull consumers at 0 and the pull path never returns for them) and trusts all ObjectMessage packages for the JDK payload types the TCK sends. Stop watching topic advisories, matching the broker's advisorySupport=false. ActiveMQConnection only populates activeTempDestinations from temp-destination advisories, and isDeleted() consults that set whenever an advisory consumer exists, so watching advisories the broker never publishes made every other connection's temporary destination read as deleted and rejected sends to a peer's temporary destination with InvalidDestinationException. Restores the temporary-destination TCK tests (temporaryQueueNotConsumableTest, temporaryTopicNotConsumableTest, temporaryTopicConnectionClosesTest, createTemporayQueueTest). ts.jtx documents two pre-existing non-compliances unrelated to shared subscriptions: the anonymous-producer xUnsupportedOperationException tests and queueReceiveTests (dispatch to consumers on never-started connections).
…destination bypass
setDeliveryDelay/getDeliveryDelay returned a fixed zero on both
MessageProducer and JMSProducer, so JMS 2.0 delivery delay was
unavailable even though the wire already carried the field: Message
gained deliveryTime with OpenWire v13, and ActiveMQMessage already
implements getJMSDeliveryTime/setJMSDeliveryTime over it.
Client: the producer stores the delay and ActiveMQSession stamps an
absolute JMSDeliveryTime of send time + delay. Absolute rather than
relative so a delay survives a hop between brokers without accumulating
drift. Per JMS 2.0 time-to-live now runs from the delivery time rather
than the send time, so a delayed message still gets its full lifetime;
with no delay configured this is identical to the previous behaviour.
Broker: SchedulerBroker translates the absolute delivery time into a
relative delay and reuses the existing scheduler, so there is one
scheduling engine and no new store. A delivery time in the past, or none
at all, delivers immediately. The explicit AMQ_SCHEDULED_* properties keep
precedence and are considered first, so everything predating delivery
delay behaves exactly as before. Messages delivered by the scheduler carry
a scheduledJobId marker that stops property-driven schedules from being
re-scheduled forever when a consumer re-sends the message; that guard is
kept for property-driven schedules only, so a re-sent message with a fresh
future JMSDeliveryTime is delayed again (the producer recomputes the
delivery time on every send).
Two ways to bypass the delay, for when a broker is only a hop and the
target should serve the wait instead:
- broker-wide: schedulerSupport=false. Already structural, since
SchedulerBroker is only installed when the scheduler is enabled.
- per-destination: the new PolicyEntry.deliveryDelayEnabled (default
true). Scoped to the JMS delivery delay only; AMQ_SCHEDULED_* messages
are still scheduled.
Either way the message keeps its JMSDeliveryTime, so a downstream broker
can serve the remaining delay. The destination policy is consulted only
once a delay is known to be present, keeping ordinary sends off that path.
The job store marshals at storeOpenWireVersion, which defaults to 11 and
so predates the v13 deliveryTime field; the delivery time is carried
across the store as a property and restored on delivery, so a scheduled
message reports the delivery time the sender asked for regardless of the
store version in use. A non-persistent broker uses the in-memory job
scheduler store; pinned by DeliveryDelayNonPersistentBrokerTest.
The setDeliveryDelay stopgap tests in ActiveMQJMS2ContextTest now assert
the set/get round trip.
… delay JMS 2.0 delivery delay is implemented by the scheduler broker, which the embedded TCK broker never enabled, so delayed messages were delivered immediately (six delivery-delay TCK failures). The non-persistent broker uses the in-memory job scheduler store.
…oducer proxy InboundMessageProducerProxy rejected setDeliveryDelay and reported a fixed zero from getDeliveryDelay while delivery delay was unimplemented. Pass both through to the wrapped MessageProducer now that the client supports them.
mattrpav
force-pushed
the
amq-gh-1717-delivery-delay
branch
from
September 22, 2026 20:03
94134ca to
140d465
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
NOTE: This PR includes the Shared Topic changes as it needs the new openwire format.
TODO: Rebase this PR once OpenWire v13 is merged.