Author: gtully
Date: Wed Feb 10 11:36:23 2010
New Revision: 908455
URL: http://svn.apache.org/viewvc?rev=908455&view=rev
Log:
merge -c 907174 https://svn.apache.org/repos/asf/activemq/trunk -
https://issues.apache.org/activemq/browse/AMQ-2590 - ensure beforeEnd is called
once per transaction so ack on rollback is avoided, add more tests
Added:
activemq/branches/activemq-5.3/activemq-core/src/test/java/org/apache/activemq/TransactionContextTest.java
- copied unchanged from r907174,
activemq/trunk/activemq-core/src/test/java/org/apache/activemq/TransactionContextTest.java
Modified:
activemq/branches/activemq-5.3/activemq-core/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java
activemq/branches/activemq-5.3/activemq-core/src/main/java/org/apache/activemq/TransactionContext.java
activemq/branches/activemq-5.3/activemq-core/src/test/java/org/apache/activemq/transport/failover/FailoverConsumerOutstandingCommitTest.java
activemq/branches/activemq-5.3/activemq-core/src/test/java/org/apache/activemq/transport/failover/FailoverTransactionTest.java
Modified:
activemq/branches/activemq-5.3/activemq-core/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java
URL:
http://svn.apache.org/viewvc/activemq/branches/activemq-5.3/activemq-core/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java?rev=908455&r1=908454&r2=908455&view=diff
==============================================================================
---
activemq/branches/activemq-5.3/activemq-core/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java
(original)
+++
activemq/branches/activemq-5.3/activemq-core/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java
Wed Feb 10 11:36:23 2010
@@ -16,6 +16,28 @@
*/
package org.apache.activemq;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+
+import javax.jms.IllegalStateException;
+import javax.jms.InvalidDestinationException;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageListener;
+import javax.jms.TransactionRolledBackException;
+
import org.apache.activemq.blob.BlobDownloader;
import org.apache.activemq.command.ActiveMQBlobMessage;
import org.apache.activemq.command.ActiveMQDestination;
@@ -29,6 +51,7 @@
import org.apache.activemq.command.MessageId;
import org.apache.activemq.command.MessagePull;
import org.apache.activemq.command.RemoveInfo;
+import org.apache.activemq.command.TransactionId;
import org.apache.activemq.management.JMSConsumerStatsImpl;
import org.apache.activemq.management.StatsCapable;
import org.apache.activemq.management.StatsImpl;
@@ -40,27 +63,6 @@
import org.apache.activemq.util.JMSExceptionSupport;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.ListIterator;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicReference;
-import javax.jms.IllegalStateException;
-import javax.jms.InvalidDestinationException;
-import javax.jms.JMSException;
-import javax.jms.Message;
-import javax.jms.MessageConsumer;
-import javax.jms.MessageListener;
-import javax.jms.TransactionRolledBackException;
/**
* A client uses a <CODE>MessageConsumer</CODE> object to receive messages
@@ -100,6 +102,14 @@
*/
public class ActiveMQMessageConsumer implements MessageAvailableConsumer,
StatsCapable, ActiveMQDispatcher {
+ @SuppressWarnings("serial")
+ class PreviouslyDeliveredMap<K, V> extends HashMap<K, V> {
+ final TransactionId transactionId;
+ public PreviouslyDeliveredMap(TransactionId transactionId) {
+ this.transactionId = transactionId;
+ }
+ }
+
private static final Log LOG =
LogFactory.getLog(ActiveMQMessageConsumer.class);
protected static final Scheduler scheduler = Scheduler.getInstance();
protected final ActiveMQSession session;
@@ -113,7 +123,7 @@
// Always walk list in reverse order.
private final LinkedList<MessageDispatch> deliveredMessages = new
LinkedList<MessageDispatch>();
// track duplicate deliveries in a transaction such that the tx integrity
can be validated
- private HashMap<MessageId, Boolean> previouslyDeliveredMessages;
+ private PreviouslyDeliveredMap<MessageId, Boolean>
previouslyDeliveredMessages;
private int deliveredCounter;
private int additionalWindowSize;
private long redeliveryDelay;
@@ -143,7 +153,6 @@
private long optimizeAckTimestamp = System.currentTimeMillis();
private long optimizeAckTimeout = 300;
private long failoverRedeliveryWaitPeriod = 0;
- private boolean rollbackInitiated;
/**
* Create a MessageConsumer
@@ -558,8 +567,7 @@
MessageDispatch md;
if (info.getPrefetchSize() == 0) {
- md = dequeue(-1); // We let the broker let us know when we
- // timeout.
+ md = dequeue(-1); // We let the broker let us know when we
timeout.
} else {
md = dequeue(timeout);
}
@@ -992,7 +1000,8 @@
}
}
if (numberNotReplayed > 0) {
- LOG.info("waiting for redelivery of " + numberNotReplayed
+ " to consumer :" + this.getConsumerId());
+ LOG.info("waiting for redelivery of " + numberNotReplayed
+ " in transaction: "
+ + previouslyDeliveredMessages.transactionId + ",
to consumer :" + this.getConsumerId());
try {
Thread.sleep(Math.max(500,
failoverRedeliveryWaitPeriod/4));
} catch (InterruptedException outOfhere) {
@@ -1008,12 +1017,6 @@
*/
private void rollbackOnFailedRecoveryRedelivery() throws JMSException {
if (previouslyDeliveredMessages != null) {
- if (rollbackInitiated) {
- // second call from rollback, nothing more to do
- // REVISIT - should beforeEnd be called again by transaction
context?
- rollbackInitiated = false;
- return;
- }
// if any previously delivered messages was not re-delivered,
transaction is invalid and must rollback
// as messages have been dispatched else where.
int numberNotReplayed = 0;
@@ -1021,20 +1024,20 @@
if (!entry.getValue()) {
numberNotReplayed++;
if (LOG.isDebugEnabled()) {
- LOG.debug("previously delivered message has not been
replayed in transaction, id: " + entry.getKey());
+ LOG.debug("previously delivered message has not been
replayed in transaction: "
+ + previouslyDeliveredMessages.transactionId
+ + " , messageId: " + entry.getKey());
}
}
}
-
if (numberNotReplayed > 0) {
- String message = "rolling back transaction post failover
recovery. " + numberNotReplayed
+ String message = "rolling back transaction ("
+ + previouslyDeliveredMessages.transactionId + ") post
failover recovery. " + numberNotReplayed
+ " previously delivered message(s) not replayed to
consumer: " + this.getConsumerId();
LOG.warn(message);
- rollbackInitiated = true;
- throw new TransactionRolledBackException(message);
+ throw new TransactionRolledBackException(message);
}
}
-
}
void acknowledge(MessageDispatch md) throws JMSException {
@@ -1157,7 +1160,6 @@
removeFromDeliveredMessages(entry.getKey());
}
}
- rollbackInitiated = false;
clearPreviouslyDelivered();
}
}
@@ -1166,9 +1168,9 @@
* called with deliveredMessages locked
*/
private void removeFromDeliveredMessages(MessageId key) {
- ListIterator<MessageDispatch> iterator =
deliveredMessages.listIterator(deliveredMessages.size());
- while (iterator.hasPrevious()) {
- MessageDispatch candidate = iterator.previous();
+ Iterator<MessageDispatch> iterator = deliveredMessages.iterator();
+ while (iterator.hasNext()) {
+ MessageDispatch candidate = iterator.next();
if (key.equals(candidate.getMessage().getMessageId())) {
session.connection.rollbackDuplicate(this,
candidate.getMessage());
iterator.remove();
@@ -1234,13 +1236,15 @@
if (previouslyDeliveredMessages != null) {
previouslyDeliveredMessages.put(md.getMessage().getMessageId(), true);
} else {
- // existing transaction gone but still a
duplicate!, lets mark as poison ftm,
- // possibly could allow redelivery..
+ // delivery while pending redelivery to
another consumer on the same connection
+ // not waiting for redelivery will help
here
needsPoisonAck = true;
}
}
if (needsPoisonAck) {
- LOG.warn("acking as poison, duplicate
transacted delivery but no recovering transaction for: " + md);
+ LOG.warn("acking duplicate delivery as poison,
redelivery must be pending to another"
+ + " consumer on this connection,
failoverRedeliveryWaitPeriod="
+ + failoverRedeliveryWaitPeriod + ".
Message: " + md);
MessageAck poisonAck = new MessageAck(md,
MessageAck.POSION_ACK_TYPE, 1);
poisonAck.setFirstMessageId(md.getMessage().getMessageId());
session.sendAck(poisonAck);
@@ -1271,7 +1275,7 @@
LOG.debug(getConsumerId() + " tracking
existing transacted delivered list (" + deliveredMessages.size() + ") on
transport interrupt");
}
if (previouslyDeliveredMessages == null) {
- previouslyDeliveredMessages = new
HashMap<MessageId, Boolean>();
+ previouslyDeliveredMessages = new
PreviouslyDeliveredMap<MessageId,
Boolean>(session.getTransactionContext().getTransactionId());
}
for (MessageDispatch delivered :
deliveredMessages) {
previouslyDeliveredMessages.put(delivered.getMessage().getMessageId(), false);
Modified:
activemq/branches/activemq-5.3/activemq-core/src/main/java/org/apache/activemq/TransactionContext.java
URL:
http://svn.apache.org/viewvc/activemq/branches/activemq-5.3/activemq-core/src/main/java/org/apache/activemq/TransactionContext.java?rev=908455&r1=908454&r2=908455&view=diff
==============================================================================
---
activemq/branches/activemq-5.3/activemq-core/src/main/java/org/apache/activemq/TransactionContext.java
(original)
+++
activemq/branches/activemq-5.3/activemq-core/src/main/java/org/apache/activemq/TransactionContext.java
Wed Feb 10 11:36:23 2010
@@ -79,6 +79,7 @@
private Xid associatedXid;
private TransactionId transactionId;
private LocalTransactionEventListener localTransactionEventListener;
+ private int beforeEndIndex;
public TransactionContext(ActiveMQConnection connection) {
this.connection = connection;
@@ -174,8 +175,8 @@
int size = synchronizations.size();
try {
- for (int i = 0; i < size; i++) {
- synchronizations.get(i).beforeEnd();
+ for (;beforeEndIndex < size;) {
+ synchronizations.get(beforeEndIndex++).beforeEnd();
}
} catch (JMSException e) {
throw e;
@@ -206,6 +207,7 @@
if (transactionId == null) {
synchronizations = null;
+ beforeEndIndex = 0;
this.transactionId = new LocalTransactionId(connectionId,
localTransactionIdGenerator.getNextSequenceId());
TransactionInfo info = new TransactionInfo(getConnectionId(),
transactionId, TransactionInfo.BEGIN);
this.connection.ensureConnectionInfoSent();
@@ -341,6 +343,7 @@
// associate
synchronizations = null;
+ beforeEndIndex = 0;
setXid(xid);
}
Modified:
activemq/branches/activemq-5.3/activemq-core/src/test/java/org/apache/activemq/transport/failover/FailoverConsumerOutstandingCommitTest.java
URL:
http://svn.apache.org/viewvc/activemq/branches/activemq-5.3/activemq-core/src/test/java/org/apache/activemq/transport/failover/FailoverConsumerOutstandingCommitTest.java?rev=908455&r1=908454&r2=908455&view=diff
==============================================================================
---
activemq/branches/activemq-5.3/activemq-core/src/test/java/org/apache/activemq/transport/failover/FailoverConsumerOutstandingCommitTest.java
(original)
+++
activemq/branches/activemq-5.3/activemq-core/src/test/java/org/apache/activemq/transport/failover/FailoverConsumerOutstandingCommitTest.java
Wed Feb 10 11:36:23 2010
@@ -287,7 +287,7 @@
Message msg = testConsumer.receive(5000);
assertNotNull(msg);
- // restart with out standing delivered message
+ // restart with outstanding delivered message
broker.stop();
broker.waitUntilStopped();
broker = createBroker(false);
Modified:
activemq/branches/activemq-5.3/activemq-core/src/test/java/org/apache/activemq/transport/failover/FailoverTransactionTest.java
URL:
http://svn.apache.org/viewvc/activemq/branches/activemq-5.3/activemq-core/src/test/java/org/apache/activemq/transport/failover/FailoverTransactionTest.java?rev=908455&r1=908454&r2=908455&view=diff
==============================================================================
---
activemq/branches/activemq-5.3/activemq-core/src/test/java/org/apache/activemq/transport/failover/FailoverTransactionTest.java
(original)
+++
activemq/branches/activemq-5.3/activemq-core/src/test/java/org/apache/activemq/transport/failover/FailoverTransactionTest.java
Wed Feb 10 11:36:23 2010
@@ -16,9 +16,11 @@
*/
package org.apache.activemq.transport.failover;
+import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
import java.io.File;
import java.io.IOException;
@@ -291,8 +293,8 @@
@Test
public void testFailoverConsumerAckLost() throws Exception {
- // as failure depends on hash order, do a few times
- for (int i=0; i<4; i++) {
+ // as failure depends on hash order of state tracker recovery, do a
few times
+ for (int i=0; i<3; i++) {
try {
doTestFailoverConsumerAckLost(i);
} finally {
@@ -482,14 +484,14 @@
broker.stop();
broker = createBroker(false);
- // use empty jdbc store so that default wait for redeliveries will
timeout after failover
+ // use empty jdbc store so that default wait(0) for redeliveries will
timeout after failover
setPersistenceAdapter(1);
broker.start();
try {
consumerSession.commit();
- } catch (JMSException expectedRolledback) {
- assertTrue(expectedRolledback instanceof
TransactionRolledBackException);
+ fail("expected transaciton rolledback ex");
+ } catch (TransactionRolledBackException expected) {
}
broker.stop();
@@ -550,6 +552,65 @@
connection.close();
}
+
+ @Test
+ public void testPoisonOnDeliveryWhilePending() throws Exception {
+ LOG.info("testWaitForMissingRedeliveries()");
+ broker = createBroker(true);
+ broker.start();
+ ActiveMQConnectionFactory cf = new
ActiveMQConnectionFactory("failover:(" + url +
")?jms.consumerFailoverRedeliveryWaitPeriod=10000");
+ Connection connection = cf.createConnection();
+ connection.start();
+ final Session producerSession = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
+ final Queue destination = producerSession.createQueue(QUEUE_NAME +
"?consumer.prefetchSize=0");
+ final Session consumerSession = connection.createSession(true,
Session.SESSION_TRANSACTED);
+ MessageConsumer consumer = consumerSession.createConsumer(destination);
+
+ produceMessage(producerSession, destination);
+ Message msg = consumer.receive(20000);
+ if (msg == null) {
+ AutoFailTestSupport.dumpAllThreads("missing-");
+ }
+ assertNotNull("got message just produced", msg);
+
+ broker.stop();
+ broker = createBroker(false);
+ broker.start();
+
+ final CountDownLatch commitDone = new CountDownLatch(1);
+
+
+ // with prefetch=0, it will not get redelivered as there will not be
another receive
+ // for this consumer. so it will block till it timeout with an
exception
+ // will block pending re-deliveries
+ Executors.newSingleThreadExecutor().execute(new Runnable() {
+ public void run() {
+ LOG.info("doing async commit...");
+ try {
+ consumerSession.commit();
+ } catch (JMSException ignored) {
+ commitDone.countDown();
+ }
+ }
+ });
+
+ // pull the pending message to this consumer where it will be poison
as it is a duplicate without a tx
+ MessageConsumer consumer2 =
consumerSession.createConsumer(consumerSession.createQueue(QUEUE_NAME +
"?consumer.prefetchSize=1"));
+ assertNull("consumer2 not get a message while pending to 1",
consumer2.receive(2000));
+
+ assertTrue("commit completed with ex", commitDone.await(15,
TimeUnit.SECONDS));
+ assertNull("consumer should not get rolledback and non redelivered
message", consumer.receive(5000));
+
+ // message should be in dlq
+ MessageConsumer dlqConsumer =
consumerSession.createConsumer(consumerSession.createQueue("ActiveMQ.DLQ"));
+ TextMessage dlqMessage = (TextMessage) dlqConsumer.receive(5000);
+ assertNotNull("found message in dlq", dlqMessage);
+ assertEquals("text matches", "Test message", dlqMessage.getText());
+ consumerSession.commit();
+
+ connection.close();
+ }
+
private void produceMessage(final Session producerSession, Queue
destination)
throws JMSException {
MessageProducer producer =
producerSession.createProducer(destination);
