Spring JMSとqpidを使用してAzure Service Busメッセージを消費するときにDeferredとしてマークされたメッセージ

Question

サービスバス購読からのメッセージを受信できますが、リスナーで例外が発生した場合、最終的に状態= undeliverableHere = trueがサービスバスに送信されます。サービスバス用のdocsは、amqp Modified処理をサポートしていないと言います。

メッセージは、サービスバスのDeferred状態になり、メッセージをどのようにしてアクティビティに振りかざすのかわかりません。

JMS設定：

@Bean 
public ConnectionFactory jmsConnectionFactory(MessageStoreProperties properties) throws UnsupportedEncodingException { 
    JmsConnectionFactory connectionFactory = new JmsConnectionFactory(properties.getUrlString()); 
    connectionFactory.setClientID(clientId); 
    connectionFactory.setUsername(properties.getUsername()); 
    connectionFactory.setPassword(properties.getPassword()); 
    connectionFactory.setRedeliveryPolicy(redeliveryPolicy()); 
    return new CachingConnectionFactory(connectionFactory); 
} 

@Bean 
public JmsDefaultRedeliveryPolicy redeliveryPolicy() { 
    JmsDefaultRedeliveryPolicy policy = new JmsDefaultRedeliveryPolicy(); 
    policy.setMaxRedeliveries(50); 
    return policy; 
} 

@Bean 
public JmsListenerContainerFactory topicContainerFactory(ConnectionFactory connectionFactory) { 
    DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory(); 
    factory.setConnectionFactory(connectionFactory); 
    factory.setSubscriptionDurable(true); 
    factory.setPubSubDomain(true); 
    factory.setSessionAcknowledgeMode(Session.CLIENT_ACKNOWLEDGE); 
    return factory; 
} 

リスナー：

@JmsListener(destination = "crm-customer-event/subscriptions/test-sub", containerFactory = "topicContainerFactory") 
public void receiveCustomerEvent(@Payload ExecutionContextDTO dto) { 
    logger.debug("Got payload: " + dto); 
    throw new RuntimeException("Oops"); 
} 

ここで私は、ログで見ているものです：

メッセージ転送が

[299309872:1] <- Transfer{handle=0, deliveryId=0, deliveryTag=\x97\x1c\x87\x04\xb7\xea\x86@\xb3n\xbd\x9fg\x81\x00\x11, messageFormat=0, settled=null, more=false, rcvSettleMode=null, state=null, resume=false, aborted=false, batchable=true} (16695) "\x00Sp\xc0\x0a\x05@@p......... 

例外がスローされますを開始、次にJMS rそれは

[299309872:1] -> Disposition{role=RECEIVER, first=0, last=0, settled=true, state=Modified{deliveryFailed=true, undeliverableHere=true, messageAnnotations=null}, batchable=false} 

メッセージ私が見るローカルに50回（なぜそれがこれを行うのですか？）

次のAMQPプロトコルフレームをあるedeliversた後、この最後の処分フレームがメッセージを入射させるように私には見えます遅延状態。メッセージにはまだロックトークンがあります。 TTLが経過しても、メッセージは依然としてサブスクリプションに滞っており、REST APIを介してメッセージを突き止めることはできません。私は（PUTで）それをロック解除し、削除（DELETEで）を削除しようとしました。私はまた、REST API（PeekLockと受信と削除の両方の種類）で受信しようとしましたが、それはそこにないように見えます。私は期限切れの後にデッド・レター・キューにメッセージを自動的に移動するサブスクリプションにオプションが設定されており、決して移動されません。

ackが発生するコードはhereです。ライブラリのこの部分は拡張されていないようですが、そうでない場合は別のackを返す独自の実装を作成しているようです。

どのように私は

1. にqpid /のJMSを得るのですか、これらの繰延メッセージを
2. ACK FAILEDの代わりMODIFIED_FAILED_UNDELIVERABLEアッククリアを使用してください。

Answer 1

最初に、JMS仕様では、MessageListenerコールバックから例外がスローされるのは実際にアプリケーションプログラミングエラーであると正しく記載されています。アプリケーションでこれらのエラーを処理する必要があります。

第2に、クライアントはメッセージがこのクライアントに配信されなかったことを示すために正しい処理を使用していますが、リモートはAMQP 1.0仕様に記載されているすべての処理をサポートする必要があります。仕様をより詳しく説明します。

上記を回避するには、クライアント肯定応答モードでメッセージを受け取ることができます。また、スローされた例外を傍受すると、this JIRA Issueで説明されているメカニズムを使用してメッセージ肯定応答を構成できます。

コードは、次のようになります。

とモードが定義されている

message.setIntProperty("JMS_AMQP_ACK_TYPE", 2); 
message.acknowledge(); 

：

ACCEPTED = 1; 
REJECTED = 2; 
RELEASED = 3; 
MODIFIED_FAILED = 4; 
MODIFIED_FAILED_UNDELIVERABLE = 5;