You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@activemq.apache.org by "Vladimir Polukeev (Jira)" <ji...@apache.org> on 2024/04/19 15:30:00 UTC
[jira] [Updated] (ARTEMIS-4732) Parameter "ackIndividually" at ClientConsumerImpl class breaks flushAcks() method in case of xa session state
[ https://issues.apache.org/jira/browse/ARTEMIS-4732?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Vladimir Polukeev updated ARTEMIS-4732:
---------------------------------------
Description:
I spent several days to figure out why rollback don't work at distributed transaction (two phase commit implementation). Finally, I have notice that acknowledgement has sent before my application code is invoked by jms message consumer.
I found out that reason of such behavior is boolean parameter "ackIndividually" at ClientConsumerImpl at artemis-client-core module.
I could not provide steps to reproduce, but I try to describe issue and show how I fix it locally in order to verify my proposal.
From my point of view, issue is here at ClientConsumerImpl class in artemis-core-client module:
{code:java}
private void handleRegularMessage(ClientMessageInternal message) {
// redundant code omitted
if (!ackIndividually && message.getPriority() != 4 && !message.containsProperty(ClientConsumerImpl.FORCED_DELIVERY_MESSAGE)) {
ackIndividually = true;
}
// redundant code omitted
}
{code}
It is sole place where parameter ackIndividually set to true value. And it is always set to true for every incoming message at my case.
So, after that code flow goes to JMSMessageListenerWrapper class at artemis-jms-client module. And method acknowledge is invoked on ClientMessage object.
{code:java}
public void onMessage(final ClientMessage message) {
// redundant code omitted
if (transactedOrClientAck) {
try {
message.acknowledge();
} catch (ActiveMQException e) {
// redundant code omitted
}
}
try {
// redundant code omitted
listener.onMessage(msg);
}
// redundant code omitted
}{code}
As you can see from code snippet above at transaction session acknowledge is invoked before message listener. This method invocation lead to ClientConsumerImpl class. Here is code snippet:
{noformat}
public void acknowledge(final ClientMessage message) throws ActiveMQException {
ClientMessageInternal cmi = (ClientMessageInternal) message;
if (ackIndividually) {
individualAcknowledge(message);
} else {
ackBytes += message.getEncodeSize();
if (logger.isTraceEnabled()) {
logger.trace("{}::acknowledge ackBytes={} and ackBatchSize={}, encodeSize={}", this, ackBytes, ackBatchSize, message.getEncodeSize());
}
if (ackBytes >= ackBatchSize) {
logger.trace("{}:: acknowledge acking {}", this, cmi);
doAck(cmi);
} else {
logger.trace("{}:: acknowledge setting lastAckedMessage = {}", this, cmi);
lastAckedMessage = cmi;
}
}
}{noformat}
And if parameter ackIndividually set to true code flow is not able to reach to else code block. And it is issue for me, because when transaction manager ends transaction, it invokes flushAcks() method at ClientConsumerImpl. Here is code snippet:
{noformat}
public void flushAcks() throws ActiveMQException {
if (lastAckedMessage != null) {
logger.trace("{}::FlushACK acking lastMessage::{}", this, lastAckedMessage);
doAck(lastAckedMessage);
}
}
{noformat}
Because of lastAckedMessage is not set at acknowledge method this method do nothing.
So, why is an issue?
It looks like all work fine when distributed transaction commit is success. But it works incorrectly in case of distributed transaction commit is fail, because acknowledgement has sent and message already left a message broker queue. Another words, rollback does not work, because rollback do not send back message to queue it works differently under the hood.
*Proposal to fix*
Add `!session.isXA() &&` to if condition in handleRegularMessage at ClientConsumerImpl class:
{noformat}
private void handleRegularMessage(ClientMessageInternal message) {
// redundant code omitted
// Transaction manager should manage message acknowledgement
// and ackIndividually breaks flushAcks() method in case of xa session state.
if (!session.isXA() && !ackIndividually && message.getPriority() != 4 && !message.containsProperty(ClientConsumerImpl.FORCED_DELIVERY_MESSAGE)) {
ackIndividually = true;
}
// redundant code omitted
}{noformat}
From my point of view, using ackIndividually at xa session makes no sense because message acknowledgement should be managed by transaction manager depend on transaction commit results.
was:
I spent several days to figure out why rollback don't work at distributed transaction (two phase commit implementation). Finally, I have notice that acknowledgement has sent before my application code is invoked by jms message consumer.
I found out that reason of such behavior is boolean parameter "ackIndividually" at ClientConsumerImpl at artemis-client-core module.
I could not provide steps to reproduce, but I try to describe issue and show how I fix it locally in order to verify my proposal.
From my point of view, issue is here at ClientConsumerImpl class in artemis-core-client module:
{code:java}
private void handleRegularMessage(ClientMessageInternal message) {
// redundant code omitted
if (!ackIndividually && message.getPriority() != 4 && !message.containsProperty(ClientConsumerImpl.FORCED_DELIVERY_MESSAGE)) {
ackIndividually = true;
}
// redundant code omitted
}
{code}
It is sole place where parameter ackIndividually set to true value. And it is always set to true for every incoming message at my case.
So, after that code flow goes to JMSMessageListenerWrapper class at artemis-jms-client module. And method acknowledge is invoked on ClientMessage object.
{code:java}
public void onMessage(final ClientMessage message) {
// redundant code omitted
if (transactedOrClientAck) {
try {
message.acknowledge();
} catch (ActiveMQException e) {
// redundant code omitted
}
}
try {
// redundant code omitted
listener.onMessage(msg);
}
// redundant code omitted
}{code}
As you can see from code snippet above at transaction session acknowledge is invoked before message listener. This method invocation lead to ClientConsumerImpl class. Here is code snippet:
{noformat}
public void acknowledge(final ClientMessage message) throws ActiveMQException {
ClientMessageInternal cmi = (ClientMessageInternal) message;
if (ackIndividually) {
individualAcknowledge(message);
} else {
ackBytes += message.getEncodeSize();
if (logger.isTraceEnabled()) {
logger.trace("{}::acknowledge ackBytes={} and ackBatchSize={}, encodeSize={}", this, ackBytes, ackBatchSize, message.getEncodeSize());
}
if (ackBytes >= ackBatchSize) {
logger.trace("{}:: acknowledge acking {}", this, cmi);
doAck(cmi);
} else {
logger.trace("{}:: acknowledge setting lastAckedMessage = {}", this, cmi);
lastAckedMessage = cmi;
}
}
}{noformat}
And if parameter ackIndividually set to true code flow is not able to reach to else code block. And it is issue for me, because when transaction manager ends transaction, it invokes flushAcks() method at ClientConsumerImpl. Here is code snippet:
{noformat}
public void flushAcks() throws ActiveMQException {
if (lastAckedMessage != null) {
logger.trace("{}::FlushACK acking lastMessage::{}", this, lastAckedMessage);
doAck(lastAckedMessage);
}
}
{noformat}
Because of lastAckedMessage is not set at acknowledge method this method do nothing.
So, why is an issue?
It looks like all work fine when distributed transaction commit is success. But it works incorrectly in case of distributed transaction commit is fail, because acknowledgement has sent and message already leave a message broker queue. Another words, rollback does not work, because rollback do not send back message to queue it works differently under the hood.
*Proposal to fix*
Add `!session.isXA() &&` to if condition in handleRegularMessage at ClientConsumerImpl class:
{noformat}
private void handleRegularMessage(ClientMessageInternal message) {
// redundant code omitted
// Transaction manager should manage message acknowledgement
// and ackIndividually breaks flushAcks() method in case of xa session state.
if (!session.isXA() && !ackIndividually && message.getPriority() != 4 && !message.containsProperty(ClientConsumerImpl.FORCED_DELIVERY_MESSAGE)) {
ackIndividually = true;
}
// redundant code omitted
}{noformat}
From my point of view, using ackIndividually at xa session makes no sense because message acknowledgement should be managed by transaction manager depend on transaction commit results.
> Parameter "ackIndividually" at ClientConsumerImpl class breaks flushAcks() method in case of xa session state
> --------------------------------------------------------------------------------------------------------------
>
> Key: ARTEMIS-4732
> URL: https://issues.apache.org/jira/browse/ARTEMIS-4732
> Project: ActiveMQ Artemis
> Issue Type: Bug
> Components: JMS
> Affects Versions: 2.33.0
> Reporter: Vladimir Polukeev
> Priority: Major
>
> I spent several days to figure out why rollback don't work at distributed transaction (two phase commit implementation). Finally, I have notice that acknowledgement has sent before my application code is invoked by jms message consumer.
>
> I found out that reason of such behavior is boolean parameter "ackIndividually" at ClientConsumerImpl at artemis-client-core module.
> I could not provide steps to reproduce, but I try to describe issue and show how I fix it locally in order to verify my proposal.
>
> From my point of view, issue is here at ClientConsumerImpl class in artemis-core-client module:
>
> {code:java}
> private void handleRegularMessage(ClientMessageInternal message) {
> // redundant code omitted
> if (!ackIndividually && message.getPriority() != 4 && !message.containsProperty(ClientConsumerImpl.FORCED_DELIVERY_MESSAGE)) {
> ackIndividually = true;
> }
>
> // redundant code omitted
> }
> {code}
> It is sole place where parameter ackIndividually set to true value. And it is always set to true for every incoming message at my case.
> So, after that code flow goes to JMSMessageListenerWrapper class at artemis-jms-client module. And method acknowledge is invoked on ClientMessage object.
>
> {code:java}
> public void onMessage(final ClientMessage message) {
> // redundant code omitted
> if (transactedOrClientAck) {
> try {
> message.acknowledge();
> } catch (ActiveMQException e) {
> // redundant code omitted
> }
> }
> try {
> // redundant code omitted
> listener.onMessage(msg);
> }
> // redundant code omitted
> }{code}
> As you can see from code snippet above at transaction session acknowledge is invoked before message listener. This method invocation lead to ClientConsumerImpl class. Here is code snippet:
>
> {noformat}
> public void acknowledge(final ClientMessage message) throws ActiveMQException {
> ClientMessageInternal cmi = (ClientMessageInternal) message;
> if (ackIndividually) {
> individualAcknowledge(message);
> } else {
> ackBytes += message.getEncodeSize();
> if (logger.isTraceEnabled()) {
> logger.trace("{}::acknowledge ackBytes={} and ackBatchSize={}, encodeSize={}", this, ackBytes, ackBatchSize, message.getEncodeSize());
> }
> if (ackBytes >= ackBatchSize) {
> logger.trace("{}:: acknowledge acking {}", this, cmi);
> doAck(cmi);
> } else {
> logger.trace("{}:: acknowledge setting lastAckedMessage = {}", this, cmi);
> lastAckedMessage = cmi;
> }
> }
> }{noformat}
> And if parameter ackIndividually set to true code flow is not able to reach to else code block. And it is issue for me, because when transaction manager ends transaction, it invokes flushAcks() method at ClientConsumerImpl. Here is code snippet:
>
> {noformat}
> public void flushAcks() throws ActiveMQException {
> if (lastAckedMessage != null) {
> logger.trace("{}::FlushACK acking lastMessage::{}", this, lastAckedMessage);
> doAck(lastAckedMessage);
> }
> }
> {noformat}
> Because of lastAckedMessage is not set at acknowledge method this method do nothing.
> So, why is an issue?
>
> It looks like all work fine when distributed transaction commit is success. But it works incorrectly in case of distributed transaction commit is fail, because acknowledgement has sent and message already left a message broker queue. Another words, rollback does not work, because rollback do not send back message to queue it works differently under the hood.
>
>
>
>
> *Proposal to fix*
> Add `!session.isXA() &&` to if condition in handleRegularMessage at ClientConsumerImpl class:
> {noformat}
> private void handleRegularMessage(ClientMessageInternal message) {
> // redundant code omitted
> // Transaction manager should manage message acknowledgement
> // and ackIndividually breaks flushAcks() method in case of xa session state.
> if (!session.isXA() && !ackIndividually && message.getPriority() != 4 && !message.containsProperty(ClientConsumerImpl.FORCED_DELIVERY_MESSAGE)) {
> ackIndividually = true;
> }
>
> // redundant code omitted
> }{noformat}
> From my point of view, using ackIndividually at xa session makes no sense because message acknowledgement should be managed by transaction manager depend on transaction commit results.
>
>
>
--
This message was sent by Atlassian Jira
(v8.20.10#820010)