You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by gt...@apache.org on 2021/01/29 09:56:39 UTC
[activemq] branch master updated: AMQ-8131 - track unmatched acks
in the acks table
This is an automated email from the ASF dual-hosted git repository.
gtully pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/activemq.git
The following commit(s) were added to refs/heads/master by this push:
new e1b3204 AMQ-8131 - track unmatched acks in the acks table
e1b3204 is described below
commit e1b3204407e0b51f0811083e77336fd71ad4c500
Author: gtully <ga...@gmail.com>
AuthorDate: Thu Jan 28 14:46:29 2021 +0000
AMQ-8131 - track unmatched acks in the acks table
---
.../activemq/store/jdbc/JDBCTopicMessageStore.java | 6 --
.../usecases/DurableSubscriptionTestSupport.java | 6 +-
.../usecases/JDBCDurableSubscriptionTest.java | 84 +++++++++++++++++++++-
3 files changed, 86 insertions(+), 10 deletions(-)
diff --git a/activemq-jdbc-store/src/main/java/org/apache/activemq/store/jdbc/JDBCTopicMessageStore.java b/activemq-jdbc-store/src/main/java/org/apache/activemq/store/jdbc/JDBCTopicMessageStore.java
index a857bcf..ccbd5c2 100644
--- a/activemq-jdbc-store/src/main/java/org/apache/activemq/store/jdbc/JDBCTopicMessageStore.java
+++ b/activemq-jdbc-store/src/main/java/org/apache/activemq/store/jdbc/JDBCTopicMessageStore.java
@@ -71,12 +71,6 @@ public class JDBCTopicMessageStore extends JDBCMessageStore implements TopicMess
@Override
public void acknowledge(ConnectionContext context, String clientId, String subscriptionName, MessageId messageId, MessageAck ack) throws IOException {
- if (ack != null && ack.isUnmatchedAck()) {
- if (LOG.isTraceEnabled()) {
- LOG.trace("ignoring unmatched selector ack for: " + messageId + ", cleanup will get to this message after subsequent acks.");
- }
- return;
- }
TransactionContext c = persistenceAdapter.getTransactionContext(context);
try {
long[] res = getCachedStoreSequenceId(c, destination, messageId);
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/DurableSubscriptionTestSupport.java b/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/DurableSubscriptionTestSupport.java
index b610f8c..0a96072 100644
--- a/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/DurableSubscriptionTestSupport.java
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/DurableSubscriptionTestSupport.java
@@ -39,11 +39,11 @@ import org.apache.activemq.store.PersistenceAdapter;
*/
public abstract class DurableSubscriptionTestSupport extends TestSupport {
- private Connection connection;
+ protected Connection connection;
private Session session;
private TopicSubscriber consumer;
private MessageProducer producer;
- private BrokerService broker;
+ protected BrokerService broker;
@Override
protected ActiveMQConnectionFactory createConnectionFactory() throws Exception {
@@ -525,7 +525,7 @@ public abstract class DurableSubscriptionTestSupport extends TestSupport {
return DeliveryMode.PERSISTENT;
}
- private void assertTextMessageEquals(String string, Message message) throws JMSException {
+ void assertTextMessageEquals(String string, Message message) throws JMSException {
assertNotNull("Message was null", message);
assertTrue("Message is not a TextMessage", message instanceof TextMessage);
assertEquals(string, ((TextMessage) message).getText());
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/JDBCDurableSubscriptionTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/JDBCDurableSubscriptionTest.java
index bc4215d..5cb336b 100644
--- a/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/JDBCDurableSubscriptionTest.java
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/JDBCDurableSubscriptionTest.java
@@ -17,10 +17,20 @@
package org.apache.activemq.usecases;
import java.io.IOException;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
import org.apache.activemq.store.PersistenceAdapter;
import org.apache.activemq.store.jdbc.JDBCPersistenceAdapter;
-import org.apache.derby.jdbc.EmbeddedDataSource;
+
+import javax.jms.DeliveryMode;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+import javax.jms.TextMessage;
+import javax.jms.Topic;
+import javax.jms.TopicSubscriber;
/**
*
@@ -33,4 +43,76 @@ public class JDBCDurableSubscriptionTest extends DurableSubscriptionTestSupport
return jdbc;
}
+ public void testUnmatchedCleanedUp() throws Exception {
+
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ Topic topic = session.createTopic("TestSelectorNoMatchCleanup");
+ TopicSubscriber consumer = session.createDurableSubscriber(topic, "sub1", "color='red'", false);
+ TopicSubscriber consumerNoMatch = session.createDurableSubscriber(topic, "sub2", "color='green'", false);
+ MessageProducer producer = session.createProducer(topic);
+ producer.setDeliveryMode(DeliveryMode.PERSISTENT);
+ connection.start();
+
+ TextMessage msg = session.createTextMessage();
+ msg.setText("Msg:1");
+ msg.setStringProperty("color", "blue");
+ producer.send(msg);
+ msg.setText("Msg:2");
+ msg.setStringProperty("color", "red");
+ producer.send(msg);
+
+ assertTextMessageEquals("Msg:2", consumer.receive(5000));
+
+ assertNull(consumerNoMatch.receiveNoWait());
+
+ // verify cleanup
+ java.sql.Connection conn = ((JDBCPersistenceAdapter) broker.getPersistenceAdapter()).getDataSource().getConnection();
+ PreparedStatement statement = conn.prepareStatement("SELECT ID FROM ACTIVEMQ_MSGS");
+ ResultSet result = statement.executeQuery();
+ printResults("MSGS", result);
+ statement.close();
+
+ statement = conn.prepareStatement("SELECT * FROM ACTIVEMQ_ACKS");
+ result = statement.executeQuery();
+ printResults("ACKS", result);
+ statement.close();
+
+ // run for each priority
+ for (int i=0; i<10; i++) {
+ ((JDBCPersistenceAdapter) broker.getPersistenceAdapter()).cleanup();
+ }
+
+ // after cleanup
+ statement = conn.prepareStatement("SELECT ID FROM ACTIVEMQ_MSGS");
+ result = statement.executeQuery();
+ printResults("MSGS-AFTER", result);
+ statement.close();
+
+ statement = conn.prepareStatement("SELECT * FROM ACTIVEMQ_ACKS");
+ result = statement.executeQuery();
+ printResults("ACKS-AFTER", result);
+ statement.close();
+
+
+ // verify empty
+ statement = conn.prepareStatement("SELECT * FROM ACTIVEMQ_MSGS");
+ result = statement.executeQuery();
+ assertFalse(result.next());
+
+ conn.close();
+ }
+
+ private void printResults(String detail, ResultSet result) throws SQLException {
+ System.out.println("**" + detail + "**");
+ ResultSetMetaData resultSetMetaData = result.getMetaData();
+ int columnsNumber = resultSetMetaData.getColumnCount();
+ while (result.next()) {
+ for (int i = 1; i <= columnsNumber; i++) {
+ if (i > 1) System.out.print(", ");
+ String columnValue = result.getString(i);
+ System.out.print(columnValue + " " + resultSetMetaData.getColumnName(i));
+ }
+ System.out.println();
+ }
+ }
}