You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by ma...@apache.org on 2024/07/16 13:08:31 UTC
(activemq) branch main updated (4a0bb48881 -> 5309352e12)
This is an automated email from the ASF dual-hosted git repository.
mattrpav pushed a change to branch main
in repository https://gitbox.apache.org/repos/asf/activemq.git
from 4a0bb48881 Merge pull request #1257 from mattrpav/AMQ-9531-b
new dad947fe4a [AMQ-9484] Support exporting kahadb messages from a queue with an offset
new 5309352e12 [AMQ-9533] Add Backup command that uses activemq-protobuf (to replace export)
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../org/apache/activemq/store/MessageStore.java | 2 +
.../apache/activemq/store/ProxyMessageStore.java | 5 +
.../activemq/store/memory/MemoryMessageStore.java | 26 +++
activemq-console/pom.xml | 15 +-
...eExportCommand.java => StoreBackupCommand.java} | 17 +-
.../console/command/StoreExportCommand.java | 3 -
...StreamManager.java => BackupStreamManager.java} | 45 ++---
.../store/{StoreExporter.java => StoreBackup.java} | 125 ++++++++----
.../src/main/{proto => protobuf}/data.proto | 2 +-
.../org.apache.activemq.console.command.Command | 1 +
.../activemq/store/jdbc/JDBCMessageStore.java | 13 ++
.../apache/activemq/store/kahadb/KahaDBStore.java | 41 ++++
.../activemq/store/kahadb/TempKahaDBStore.java | 31 +++
.../region/cursors/StoreQueueCursorOrderTest.java | 5 +
.../kahadb/KahaDBOffsetRecoveryListenerTest.java | 221 +++++++++++++++++++++
15 files changed, 471 insertions(+), 81 deletions(-)
copy activemq-console/src/main/java/org/apache/activemq/console/command/{StoreExportCommand.java => StoreBackupCommand.java} (75%)
copy activemq-console/src/main/java/org/apache/activemq/console/command/store/{ExportStreamManager.java => BackupStreamManager.java} (64%)
copy activemq-console/src/main/java/org/apache/activemq/console/command/store/{StoreExporter.java => StoreBackup.java} (75%)
copy activemq-console/src/main/{proto => protobuf}/data.proto (96%)
create mode 100644 activemq-unit-tests/src/test/java/org/apache/activemq/store/kahadb/KahaDBOffsetRecoveryListenerTest.java
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@activemq.apache.org
For additional commands, e-mail: commits-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
(activemq) 01/02: [AMQ-9484] Support exporting kahadb messages from a queue with an offset
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mattrpav pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/activemq.git
commit dad947fe4aca288256733bbb1bc442e593f26196
Author: Matt Pavlovich <ma...@apache.org>
AuthorDate: Tue Apr 23 11:43:09 2024 -0500
[AMQ-9484] Support exporting kahadb messages from a queue with an offset
(cherry picked from commit 1a1b42f0c9bd784b8c8aee4f42deab140d7480e3)
---
.../org/apache/activemq/store/MessageStore.java | 2 +
.../apache/activemq/store/ProxyMessageStore.java | 5 +
.../activemq/store/memory/MemoryMessageStore.java | 26 +++
.../activemq/store/jdbc/JDBCMessageStore.java | 13 ++
.../apache/activemq/store/kahadb/KahaDBStore.java | 41 ++++
.../activemq/store/kahadb/TempKahaDBStore.java | 31 +++
.../region/cursors/StoreQueueCursorOrderTest.java | 5 +
.../kahadb/KahaDBOffsetRecoveryListenerTest.java | 221 +++++++++++++++++++++
8 files changed, 344 insertions(+)
diff --git a/activemq-broker/src/main/java/org/apache/activemq/store/MessageStore.java b/activemq-broker/src/main/java/org/apache/activemq/store/MessageStore.java
index aee619a27a..e35327f848 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/store/MessageStore.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/store/MessageStore.java
@@ -178,6 +178,8 @@ public interface MessageStore extends Service {
void recoverNextMessages(int maxReturned, MessageRecoveryListener listener) throws Exception;
+ void recoverNextMessages(int offset, int maxReturned, MessageRecoveryListener listener) throws Exception;
+
void dispose(ConnectionContext context);
/**
diff --git a/activemq-broker/src/main/java/org/apache/activemq/store/ProxyMessageStore.java b/activemq-broker/src/main/java/org/apache/activemq/store/ProxyMessageStore.java
index cd319a65d3..a4fb4be5b8 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/store/ProxyMessageStore.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/store/ProxyMessageStore.java
@@ -110,6 +110,11 @@ public class ProxyMessageStore implements MessageStore {
delegate.recoverNextMessages(maxReturned, listener);
}
+ @Override
+ public void recoverNextMessages(int offset, int maxReturned, MessageRecoveryListener listener) throws Exception {
+ delegate.recoverNextMessages(offset, maxReturned, listener);
+ }
+
@Override
public void resetBatching() {
delegate.resetBatching();
diff --git a/activemq-broker/src/main/java/org/apache/activemq/store/memory/MemoryMessageStore.java b/activemq-broker/src/main/java/org/apache/activemq/store/memory/MemoryMessageStore.java
index 953c83e19a..f0857fb960 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/store/memory/MemoryMessageStore.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/store/memory/MemoryMessageStore.java
@@ -130,6 +130,32 @@ public class MemoryMessageStore extends AbstractMessageStore {
}
}
+ @Override
+ public void recoverNextMessages(int offset, int maxReturned, MessageRecoveryListener listener) throws Exception {
+ synchronized (messageTable) {
+ boolean pastLackBatch = lastBatchId == null;
+ int position = 0;
+ for (Map.Entry<MessageId, Message> entry : messageTable.entrySet()) {
+ if(offset > 0 && offset > position) {
+ position++;
+ continue;
+ }
+ if (pastLackBatch) {
+ Object msg = entry.getValue();
+ lastBatchId = entry.getKey();
+ if (msg.getClass() == MessageId.class) {
+ listener.recoverMessageReference((MessageId) msg);
+ } else {
+ listener.recoverMessage((Message) msg);
+ }
+ } else {
+ pastLackBatch = entry.getKey().equals(lastBatchId);
+ }
+ position++;
+ }
+ }
+ }
+
@Override
public void resetBatching() {
lastBatchId = null;
diff --git a/activemq-jdbc-store/src/main/java/org/apache/activemq/store/jdbc/JDBCMessageStore.java b/activemq-jdbc-store/src/main/java/org/apache/activemq/store/jdbc/JDBCMessageStore.java
index 8adc2f78ee..70ddb7ab1e 100644
--- a/activemq-jdbc-store/src/main/java/org/apache/activemq/store/jdbc/JDBCMessageStore.java
+++ b/activemq-jdbc-store/src/main/java/org/apache/activemq/store/jdbc/JDBCMessageStore.java
@@ -398,6 +398,19 @@ public class JDBCMessageStore extends AbstractMessageStore {
}
+ /**
+ * @param offset
+ * @param maxReturned
+ * @param listener
+ * @throws Exception
+ * @see org.apache.activemq.store.MessageStore#recoverNextMessages(int,
+ * org.apache.activemq.store.MessageRecoveryListener)
+ */
+ @Override
+ public void recoverNextMessages(int offset, int maxReturned, final MessageRecoveryListener listener) throws Exception {
+ throw new UnsupportedOperationException("recoverNextMesage(offset,maxReturned,listener) is not supported.");
+ }
+
public void trackRollbackAck(Message message) {
synchronized (rolledBackAcks) {
rolledBackAcks.put((Long)message.getMessageId().getEntryLocator(), message);
diff --git a/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/KahaDBStore.java b/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/KahaDBStore.java
index 50224a5dfa..31a86bcae0 100644
--- a/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/KahaDBStore.java
+++ b/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/KahaDBStore.java
@@ -732,6 +732,47 @@ public class KahaDBStore extends MessageDatabase implements PersistenceAdapter,
}
}
+ @Override
+ public void recoverNextMessages(final int offset, final int maxReturned, final MessageRecoveryListener listener) throws Exception {
+ indexLock.writeLock().lock();
+ try {
+ pageFile.tx().execute(new Transaction.Closure<Exception>() {
+ @Override
+ public void execute(Transaction tx) throws Exception {
+ StoredDestination sd = getStoredDestination(dest, tx);
+ Entry<Long, MessageKeys> entry = null;
+ int position = 0;
+ int counter = recoverRolledBackAcks(destination.getPhysicalName(), sd, tx, maxReturned, listener);
+ Set ackedAndPrepared = ackedAndPreparedMap.get(destination.getPhysicalName());
+ for (Iterator<Entry<Long, MessageKeys>> iterator = sd.orderIndex.iterator(tx); iterator.hasNext(); ) {
+ entry = iterator.next();
+
+ if (ackedAndPrepared != null && ackedAndPrepared.contains(entry.getValue().messageId)) {
+ continue;
+ }
+
+ if(offset > 0 && offset > position) {
+ position++;
+ continue;
+ }
+
+ Message msg = loadMessage(entry.getValue().location);
+ msg.getMessageId().setFutureOrSequenceLong(entry.getKey());
+ listener.recoverMessage(msg);
+ counter++;
+ position++;
+ if (counter >= maxReturned || !listener.canRecoveryNextMessage()) {
+ break;
+ }
+ }
+ sd.orderIndex.stoppedIterating();
+ }
+ });
+ } finally {
+ indexLock.writeLock().unlock();
+ }
+ }
+
protected int recoverRolledBackAcks(String recoveredTxStateMapKey, StoredDestination sd, Transaction tx, int maxReturned, MessageRecoveryListener listener) throws Exception {
int counter = 0;
String id;
diff --git a/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/TempKahaDBStore.java b/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/TempKahaDBStore.java
index 7048b09a44..7835a1b7b6 100644
--- a/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/TempKahaDBStore.java
+++ b/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/TempKahaDBStore.java
@@ -241,6 +241,37 @@ public class TempKahaDBStore extends TempMessageDatabase implements PersistenceA
}
}
+ @Override
+ public void recoverNextMessages(final int offset, final int maxReturned, final MessageRecoveryListener listener) throws Exception {
+ synchronized(indexMutex) {
+ pageFile.tx().execute(new Transaction.Closure<Exception>(){
+ @Override
+ public void execute(Transaction tx) throws Exception {
+ StoredDestination sd = getStoredDestination(dest, tx);
+ Entry<Long, MessageRecord> entry=null;
+ int counter = 0;
+ int position = 0;
+ for (Iterator<Entry<Long, MessageRecord>> iterator = sd.orderIndex.iterator(tx, cursorPos); iterator.hasNext();) {
+ entry = iterator.next();
+ if(offset > 0 && offset > position) {
+ position++;
+ continue;
+ }
+ listener.recoverMessage( (Message) wireFormat.unmarshal(entry.getValue().data ) );
+ counter++;
+ position++;
+ if( counter >= maxReturned ) {
+ break;
+ }
+ }
+ if( entry!=null ) {
+ cursorPos = entry.getKey()+1;
+ }
+ }
+ });
+ }
+ }
+
@Override
public void resetBatching() {
cursorPos=0;
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/broker/region/cursors/StoreQueueCursorOrderTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/broker/region/cursors/StoreQueueCursorOrderTest.java
index 5a1ab90d58..cbb1579b57 100644
--- a/activemq-unit-tests/src/test/java/org/apache/activemq/broker/region/cursors/StoreQueueCursorOrderTest.java
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/broker/region/cursors/StoreQueueCursorOrderTest.java
@@ -510,6 +510,11 @@ public class StoreQueueCursorOrderTest {
}
}
+ @Override
+ public void recoverNextMessages(int offset, int maxReturned, MessageRecoveryListener listener) throws Exception {
+
+ }
+
@Override
public void setBatch(MessageId message) {
batch.set((Long)message.getFutureOrSequenceLong());
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/store/kahadb/KahaDBOffsetRecoveryListenerTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/store/kahadb/KahaDBOffsetRecoveryListenerTest.java
new file mode 100644
index 0000000000..9ea4b68a84
--- /dev/null
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/store/kahadb/KahaDBOffsetRecoveryListenerTest.java
@@ -0,0 +1,221 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.activemq.store.kahadb;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.LinkedList;
+import java.util.List;
+
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.command.ActiveMQQueue;
+import org.apache.activemq.command.Message;
+import org.apache.activemq.command.MessageId;
+import org.apache.activemq.store.MessageRecoveryListener;
+import org.apache.activemq.store.MessageStore;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import jakarta.jms.Connection;
+import jakarta.jms.JMSException;
+import jakarta.jms.MessageConsumer;
+import jakarta.jms.MessageProducer;
+import jakarta.jms.Session;
+import jakarta.jms.TextMessage;
+
+import static org.junit.Assert.assertEquals;
+
+public class KahaDBOffsetRecoveryListenerTest {
+
+ protected BrokerService brokerService = null;
+ protected KahaDBStore kaha = null;
+
+ @Before
+ public void beforeEach() throws Exception {
+
+ }
+
+ @After
+ public void afterEach() {
+ brokerService = null;
+ kaha = null;
+ }
+
+ protected BrokerService createBroker(KahaDBStore kaha) throws Exception {
+ BrokerService broker = new BrokerService();
+ broker.setUseJmx(false);
+ broker.setPersistenceAdapter(kaha);
+ broker.start();
+ broker.waitUntilStarted(10_000l);
+ return broker;
+ }
+
+ private KahaDBStore createStore(boolean delete) throws IOException {
+ KahaDBStore kaha = new KahaDBStore();
+ kaha.setDirectory(new File("target/activemq-data/kahadb-recovery-tests"));
+ if( delete ) {
+ kaha.deleteAllMessages();
+ }
+ return kaha;
+ }
+
+ protected void runOffsetTest(int sendCount, int expectedMessageCount, int recoverOffset, int recoverCount, int expectedRecoverCount, int expectedRecoverIndex, String queueName) throws Exception {
+ kaha = createStore(true);
+ kaha.setJournalMaxFileLength(1024*100);
+ brokerService = createBroker(kaha);
+ sendMessages(sendCount, queueName);
+ brokerService.stop();
+ brokerService.waitUntilStopped();
+
+ TestMessageRecoveryListener testMessageRecoveryListener = new TestMessageRecoveryListener();
+ kaha = createStore(false);
+ kaha.start();
+ MessageStore messageStore = kaha.createQueueMessageStore(new ActiveMQQueue(queueName));
+ messageStore.start();
+ assertEquals(Integer.valueOf(expectedMessageCount), Integer.valueOf(messageStore.getMessageCount()));
+ messageStore.recoverNextMessages(recoverOffset, recoverCount, testMessageRecoveryListener);
+ messageStore.stop();
+ kaha.stop();
+
+ assertEquals(Integer.valueOf(expectedRecoverCount), Integer.valueOf(testMessageRecoveryListener.getRecoveredMessages().size()));
+
+ if(expectedRecoverIndex >= 0) {
+ assertEquals(Integer.valueOf(expectedRecoverIndex), (Integer)testMessageRecoveryListener.getRecoveredMessages().get(0).getProperty("index"));
+ }
+
+ brokerService = createBroker(kaha);
+ assertEquals(sendCount, receiveMessages(queueName));
+ }
+
+ @Test
+ public void testOffsetZero() throws Exception {
+ runOffsetTest(1_000, 1_000, 0, 1, 1, 0, "TEST.OFFSET.ZERO");
+ }
+
+ @Test
+ public void testOffsetOne() throws Exception {
+ runOffsetTest(1_000, 1_000, 1, 1, 1, 1, "TEST.OFFSET.ONE");
+ }
+
+ @Test
+ public void testOffsetLastMinusOne() throws Exception {
+ runOffsetTest(1_000, 1_000, 999, 1, 1, 999, "TEST.OFFSET.LASTMINUSONE");
+ }
+
+ @Test
+ public void testOffsetLast() throws Exception {
+ runOffsetTest(1_000, 1_000, 1_000, 1, 0, -1, "TEST.OFFSET.LAST");
+ }
+
+ @Test
+ public void testOffsetBeyondQueueSizeNoError() throws Exception {
+ runOffsetTest(1_000, 1_000, 10_000, 1, 0, -1, "TEST.OFFSET.BEYOND");
+ }
+
+ @Test
+ public void testOffsetEmptyQueue() throws Exception {
+ runOffsetTest(0, 0, 10_000, 1, 0, -1, "TEST.OFFSET.EMPTY");
+ }
+
+ private void sendMessages(int count, String queueName) throws JMSException {
+ ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory("vm://localhost");
+ cf.setUseAsyncSend(true);
+ cf.setProducerWindowSize(1024);
+ cf.setWatchTopicAdvisories(false);
+
+ Connection connection = cf.createConnection();
+ try {
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ MessageProducer producer = session.createProducer(new ActiveMQQueue(queueName));
+ for (int i = 0; i < count; i++) {
+ TextMessage textMessage = session.createTextMessage(createContent(i));
+ textMessage.setIntProperty("index", i);
+ producer.send(textMessage);
+ }
+ } finally {
+ connection.close();
+ }
+ }
+
+ private int receiveMessages(String queueName) throws JMSException {
+ int rc=0;
+ ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory("vm://localhost");
+ Connection connection = cf.createConnection();
+ try {
+ connection.start();
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ MessageConsumer messageConsumer = session.createConsumer(new ActiveMQQueue(queueName));
+ while ( messageConsumer.receive(1000) !=null ) {
+ rc++;
+ }
+ return rc;
+ } finally {
+ connection.close();
+ }
+ }
+
+ private String createContent(int i) {
+ StringBuilder sb = new StringBuilder(i+":");
+ while( sb.length() < 1024 ) {
+ sb.append("*");
+ }
+ return sb.toString();
+ }
+
+ static class TestMessageRecoveryListener implements MessageRecoveryListener {
+
+ List<MessageId> recoveredMessageIds = new LinkedList<>();
+ List<Message> recoveredMessages = new LinkedList<>();
+
+ @Override
+ public boolean hasSpace() {
+ return true;
+ }
+
+ @Override
+ public boolean isDuplicate(MessageId messageId) {
+ return recoveredMessageIds.contains(messageId);
+ }
+
+ @Override
+ public boolean recoverMessage(Message message) throws Exception {
+ if(recoveredMessages.contains(message)) {
+ return false;
+ }
+ return recoveredMessages.add(message);
+ }
+
+ @Override
+ public boolean recoverMessageReference(MessageId messageId) throws Exception {
+ if(recoveredMessageIds.contains(messageId)) {
+ return false;
+ }
+ return recoveredMessageIds.add(messageId);
+ }
+
+ public List<MessageId> getRecoveredMessageIds() {
+ return recoveredMessageIds;
+ }
+
+ public List<Message> getRecoveredMessages() {
+ return recoveredMessages;
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@activemq.apache.org
For additional commands, e-mail: commits-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
(activemq) 02/02: [AMQ-9533] Add Backup command that uses activemq-protobuf (to replace export)
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mattrpav pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/activemq.git
commit 5309352e12c1d5f05c027331f49cc1caa6a62f2f
Author: Matt Pavlovich <ma...@apache.org>
AuthorDate: Thu Jul 11 13:17:23 2024 -0500
[AMQ-9533] Add Backup command that uses activemq-protobuf (to replace export)
---
activemq-console/pom.xml | 15 +-
...eExportCommand.java => StoreBackupCommand.java} | 17 +-
.../console/command/StoreExportCommand.java | 3 -
.../console/command/store/BackupStreamManager.java | 82 ++++++
.../console/command/store/StoreBackup.java | 326 +++++++++++++++++++++
activemq-console/src/main/protobuf/data.proto | 60 ++++
.../org.apache.activemq.console.command.Command | 1 +
7 files changed, 490 insertions(+), 14 deletions(-)
diff --git a/activemq-console/pom.xml b/activemq-console/pom.xml
index 4aeab8aa8c..0158d80737 100644
--- a/activemq-console/pom.xml
+++ b/activemq-console/pom.xml
@@ -168,7 +168,20 @@
</includes>
</configuration>
</plugin>
-
+ <plugin>
+ <groupId>org.apache.activemq.protobuf</groupId>
+ <artifactId>activemq-protobuf</artifactId>
+ <configuration>
+ <sourceDirectory>src/main/protobuf</sourceDirectory>
+ </configuration>
+ <executions>
+ <execution>
+ <goals>
+ <goal>compile</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
<plugin>
<groupId>org.fusesource.hawtbuf</groupId>
<artifactId>hawtbuf-protoc</artifactId>
diff --git a/activemq-console/src/main/java/org/apache/activemq/console/command/StoreExportCommand.java b/activemq-console/src/main/java/org/apache/activemq/console/command/StoreBackupCommand.java
similarity index 75%
copy from activemq-console/src/main/java/org/apache/activemq/console/command/StoreExportCommand.java
copy to activemq-console/src/main/java/org/apache/activemq/console/command/StoreBackupCommand.java
index c53add7039..43a6f278bb 100644
--- a/activemq-console/src/main/java/org/apache/activemq/console/command/StoreExportCommand.java
+++ b/activemq-console/src/main/java/org/apache/activemq/console/command/StoreBackupCommand.java
@@ -17,16 +17,13 @@
package org.apache.activemq.console.command;
import org.apache.activemq.console.CommandContext;
-import org.apache.activemq.console.command.store.StoreExporter;
+import org.apache.activemq.console.command.store.StoreBackup;
import org.apache.activemq.console.command.store.amq.CommandLineSupport;
import java.util.Arrays;
import java.util.List;
-/**
- * @author <a href="http://hiramchirino.com">Hiram Chirino</a>
- */
-public class StoreExportCommand implements Command {
+public class StoreBackupCommand implements Command {
private CommandContext context;
@@ -37,21 +34,21 @@ public class StoreExportCommand implements Command {
@Override
public String getName() {
- return "export";
+ return "backup";
}
@Override
public String getOneLineDescription() {
- return "Exports a stopped brokers data files to an archive file";
+ return "Backup a message (or range) from a queue to an archive file";
}
@Override
public void execute(List<String> tokens) throws Exception {
- StoreExporter exporter = new StoreExporter();
- String[] remaining = CommandLineSupport.setOptions(exporter, tokens.toArray(new String[tokens.size()]));
+ StoreBackup backup = new StoreBackup();
+ String[] remaining = CommandLineSupport.setOptions(backup, tokens.toArray(new String[tokens.size()]));
if (remaining.length > 0) {
throw new Exception("Unexpected arguments: " + Arrays.asList(remaining));
}
- exporter.execute();
+ backup.execute();
}
}
diff --git a/activemq-console/src/main/java/org/apache/activemq/console/command/StoreExportCommand.java b/activemq-console/src/main/java/org/apache/activemq/console/command/StoreExportCommand.java
index c53add7039..4da756e632 100644
--- a/activemq-console/src/main/java/org/apache/activemq/console/command/StoreExportCommand.java
+++ b/activemq-console/src/main/java/org/apache/activemq/console/command/StoreExportCommand.java
@@ -23,9 +23,6 @@ import org.apache.activemq.console.command.store.amq.CommandLineSupport;
import java.util.Arrays;
import java.util.List;
-/**
- * @author <a href="http://hiramchirino.com">Hiram Chirino</a>
- */
public class StoreExportCommand implements Command {
private CommandContext context;
diff --git a/activemq-console/src/main/java/org/apache/activemq/console/command/store/BackupStreamManager.java b/activemq-console/src/main/java/org/apache/activemq/console/command/store/BackupStreamManager.java
new file mode 100644
index 0000000000..b35ce0514a
--- /dev/null
+++ b/activemq-console/src/main/java/org/apache/activemq/console/command/store/BackupStreamManager.java
@@ -0,0 +1,82 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.console.command.store;
+
+import org.apache.activemq.console.command.store.protobuf.*;
+import org.apache.activemq.console.command.store.tar.TarEntry;
+import org.apache.activemq.console.command.store.tar.TarOutputStream;
+import org.apache.activemq.protobuf.AsciiBuffer;
+import org.apache.activemq.protobuf.Buffer;
+import org.apache.activemq.protobuf.MessageBuffer;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.zip.GZIPOutputStream;
+
+public class BackupStreamManager {
+
+ private final OutputStream target;
+ private final int version;
+ TarOutputStream stream;
+
+ BackupStreamManager(OutputStream target, int version) throws IOException {
+ this.target = target;
+ this.version = version;
+ stream = new TarOutputStream(new GZIPOutputStream(target));
+ store("ver", new AsciiBuffer(""+version));
+ }
+
+
+ long seq = 0;
+
+ public void finish() throws IOException {
+ stream.close();
+ }
+
+ private void store(String ext, Buffer value) throws IOException {
+ TarEntry entry = new TarEntry(seq + "." + ext);
+ seq += 1;
+ entry.setSize(value.getLength());
+ stream.putNextEntry(entry);
+ stream.write(value.getData());
+ stream.closeEntry();
+ }
+
+ private void store(String ext, MessageBuffer<?,?> value) throws IOException {
+ TarEntry entry = new TarEntry(seq + "." + ext);
+ seq += 1;
+ entry.setSize(value.serializedSizeFramed());
+ stream.putNextEntry(entry);
+ value.writeFramed(stream);
+ stream.closeEntry();
+ }
+
+
+ public void store_queue(QueuePB value) throws IOException {
+ store("que", value.toUnframedBuffer());
+ }
+ public void store_queue_entry(QueueEntryPB value) throws IOException {
+ store("qen", value.toUnframedBuffer());
+ }
+ public void store_message(MessagePB value) throws IOException {
+ store("msg", value.toUnframedBuffer());
+ }
+ public void store_map_entry(MapEntryPB value) throws IOException {
+ store("map", value.toUnframedBuffer());
+ }
+
+}
diff --git a/activemq-console/src/main/java/org/apache/activemq/console/command/store/StoreBackup.java b/activemq-console/src/main/java/org/apache/activemq/console/command/store/StoreBackup.java
new file mode 100644
index 0000000000..026accf0eb
--- /dev/null
+++ b/activemq-console/src/main/java/org/apache/activemq/console/command/store/StoreBackup.java
@@ -0,0 +1,326 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.console.command.store;
+
+import java.io.BufferedOutputStream;
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.HashMap;
+
+import org.apache.activemq.broker.BrokerFactory;
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.command.ActiveMQDestination;
+import org.apache.activemq.command.ActiveMQQueue;
+import org.apache.activemq.command.ActiveMQTopic;
+import org.apache.activemq.command.Message;
+import org.apache.activemq.command.MessageAck;
+import org.apache.activemq.command.MessageId;
+import org.apache.activemq.command.SubscriptionInfo;
+import org.apache.activemq.command.XATransactionId;
+import org.apache.activemq.console.command.store.protobuf.MessagePB;
+import org.apache.activemq.console.command.store.protobuf.QueueEntryPB;
+import org.apache.activemq.console.command.store.protobuf.QueuePB;
+import org.apache.activemq.openwire.OpenWireFormat;
+import org.apache.activemq.store.MessageRecoveryListener;
+import org.apache.activemq.store.MessageStore;
+import org.apache.activemq.store.PersistenceAdapter;
+import org.apache.activemq.store.TopicMessageStore;
+import org.apache.activemq.store.TransactionRecoveryListener;
+import org.fusesource.hawtbuf.DataByteArrayOutputStream;
+import org.apache.activemq.protobuf.AsciiBuffer;
+import org.apache.activemq.protobuf.Buffer;
+import org.apache.activemq.protobuf.UTF8Buffer;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+public class StoreBackup {
+
+ static final int OPENWIRE_VERSION = 11;
+ static final boolean TIGHT_ENCODING = false;
+
+ URI config;
+ String filename;
+ File file;
+
+ String queue;
+ Integer offset;
+ Integer count;
+
+ private final ObjectMapper mapper = new ObjectMapper();
+ private final AsciiBuffer ds_kind = new AsciiBuffer("ds");
+ private final AsciiBuffer ptp_kind = new AsciiBuffer("ptp");
+ private final AsciiBuffer codec_id = new AsciiBuffer("openwire");
+ private final OpenWireFormat wireformat = new OpenWireFormat();
+
+ public StoreBackup() throws URISyntaxException {
+ config = new URI("xbean:activemq.xml");
+ wireformat.setCacheEnabled(false);
+ wireformat.setTightEncodingEnabled(TIGHT_ENCODING);
+ wireformat.setVersion(OPENWIRE_VERSION);
+ }
+
+ public void execute() throws Exception {
+ if (config == null) {
+ throw new Exception("required --config option missing");
+ }
+ if (filename == null) {
+ throw new Exception("required --filename option missing");
+ }
+
+ if (offset != null && count == null) {
+ throw new Exception("optional --offset and --count must be specified together");
+ }
+
+ setFile(new File(filename));
+ System.out.println("Loading: " + config);
+ BrokerFactory.setStartDefault(false); // to avoid the broker auto-starting..
+ BrokerService broker = BrokerFactory.createBroker(config);
+ BrokerFactory.resetStartDefault();
+ PersistenceAdapter store = broker.getPersistenceAdapter();
+
+ System.out.println("Starting: " + store);
+ store.start();
+ try(BufferedOutputStream fos = new BufferedOutputStream(new FileOutputStream(file))) {
+ export(store, fos);
+ } finally {
+ store.stop();
+ }
+ }
+
+ void export(PersistenceAdapter store, BufferedOutputStream fos) throws Exception {
+
+ final long[] messageKeyCounter = new long[]{0};
+ final long[] containerKeyCounter = new long[]{0};
+ final BackupStreamManager manager = new BackupStreamManager(fos, 1);
+
+ final int[] preparedTxs = new int[]{0};
+ store.createTransactionStore().recover(new TransactionRecoveryListener() {
+ @Override
+ public void recover(XATransactionId xid, Message[] addedMessages, MessageAck[] aks) {
+ preparedTxs[0] += 1;
+ }
+ });
+
+ if (preparedTxs[0] > 0) {
+ throw new Exception("Cannot export a store with prepared XA transactions. Please commit or rollback those transactions before attempting to backup.");
+ }
+
+ for (ActiveMQDestination odest : store.getDestinations()) {
+
+ if(queue != null && !queue.equals(odest.getPhysicalName())) {
+ continue;
+ }
+
+ containerKeyCounter[0]++;
+ if (odest instanceof ActiveMQQueue) {
+ ActiveMQQueue dest = (ActiveMQQueue) odest;
+ MessageStore queue = store.createQueueMessageStore(dest);
+
+ QueuePB destRecord = new QueuePB();
+ destRecord.setKey(containerKeyCounter[0]);
+ destRecord.setBindingKind(ptp_kind);
+
+ final long[] seqKeyCounter = new long[]{0};
+
+ HashMap<String, Object> jsonMap = new HashMap<String, Object>();
+ jsonMap.put("@class", "queue_destination");
+ jsonMap.put("name", dest.getQueueName());
+ String json = mapper.writeValueAsString(jsonMap);
+ System.out.println(json);
+ destRecord.setBindingData(new UTF8Buffer(json));
+ manager.store_queue(destRecord);
+
+ MessageRecoveryListener queueRecoveryListener = new MessageRecoveryListener() {
+
+ @Override
+ public boolean hasSpace() {
+ return true;
+ }
+
+ @Override
+ public boolean recoverMessageReference(MessageId ref) throws Exception {
+ return true;
+ }
+
+ @Override
+ public boolean isDuplicate(MessageId ref) {
+ return false;
+ }
+
+ @Override
+ public boolean recoverMessage(Message message) throws IOException {
+ messageKeyCounter[0]++;
+ seqKeyCounter[0]++;
+
+ MessagePB messageRecord = createMessagePB(message, messageKeyCounter[0]);
+ manager.store_message(messageRecord);
+
+ QueueEntryPB entryRecord = createQueueEntryPB(message, containerKeyCounter[0], seqKeyCounter[0], messageKeyCounter[0]);
+ manager.store_queue_entry(entryRecord);
+
+ return true;
+ }
+ };
+ if(offset != null) {
+ queue.recoverNextMessages(offset, count, queueRecoveryListener);
+ } else {
+ queue.recover(queueRecoveryListener);
+ }
+ } else if (odest instanceof ActiveMQTopic) {
+ ActiveMQTopic dest = (ActiveMQTopic) odest;
+
+ TopicMessageStore topic = store.createTopicMessageStore(dest);
+ for (SubscriptionInfo sub : topic.getAllSubscriptions()) {
+
+ QueuePB destRecord = new QueuePB();
+ destRecord.setKey(containerKeyCounter[0]);
+ destRecord.setBindingKind(ds_kind);
+
+ // TODO: use a real JSON encoder like jackson.
+ HashMap<String, Object> jsonMap = new HashMap<String, Object>();
+ jsonMap.put("@class", "dsub_destination");
+ jsonMap.put("name", sub.getClientId() + ":" + sub.getSubscriptionName());
+ HashMap<String, Object> jsonTopic = new HashMap<String, Object>();
+ jsonTopic.put("name", dest.getTopicName());
+ jsonMap.put("topics", new Object[]{jsonTopic});
+ if (sub.getSelector() != null) {
+ jsonMap.put("selector", sub.getSelector());
+ }
+ jsonMap.put("noLocal", sub.isNoLocal());
+ String json = mapper.writeValueAsString(jsonMap);
+ System.out.println(json);
+
+ destRecord.setBindingData(new UTF8Buffer(json));
+ manager.store_queue(destRecord);
+
+ final long seqKeyCounter[] = new long[]{0};
+ topic.recoverSubscription(sub.getClientId(), sub.getSubscriptionName(), new MessageRecoveryListener() {
+ @Override
+ public boolean hasSpace() {
+ return true;
+ }
+
+ @Override
+ public boolean recoverMessageReference(MessageId ref) throws Exception {
+ return true;
+ }
+
+ @Override
+ public boolean isDuplicate(MessageId ref) {
+ return false;
+ }
+
+ @Override
+ public boolean recoverMessage(Message message) throws IOException {
+ messageKeyCounter[0]++;
+ seqKeyCounter[0]++;
+
+ MessagePB messageRecord = createMessagePB(message, messageKeyCounter[0]);
+ manager.store_message(messageRecord);
+
+ QueueEntryPB entryRecord = createQueueEntryPB(message, containerKeyCounter[0], seqKeyCounter[0], messageKeyCounter[0]);
+ manager.store_queue_entry(entryRecord);
+ return true;
+ }
+ });
+
+ }
+ }
+ }
+ manager.finish();
+ }
+
+ private QueueEntryPB createQueueEntryPB(Message message, long queueKey, long queueSeq, long messageKey) {
+ QueueEntryPB entryRecord = new QueueEntryPB();
+ entryRecord.setQueueKey(queueKey);
+ entryRecord.setQueueSeq(queueSeq);
+ entryRecord.setMessageKey(messageKey);
+ entryRecord.setSize(message.getSize());
+ if (message.getExpiration() != 0) {
+ entryRecord.setExpiration(message.getExpiration());
+ }
+ if (message.getRedeliveryCounter() != 0) {
+ entryRecord.setRedeliveries(message.getRedeliveryCounter());
+ }
+ return entryRecord;
+ }
+
+ private MessagePB createMessagePB(Message message, long messageKey) throws IOException {
+ DataByteArrayOutputStream mos = new DataByteArrayOutputStream();
+ mos.writeBoolean(TIGHT_ENCODING);
+ mos.writeVarInt(OPENWIRE_VERSION);
+ wireformat.marshal(message, mos);
+
+ MessagePB messageRecord = new MessagePB();
+ messageRecord.setCodec(codec_id);
+ messageRecord.setMessageKey(messageKey);
+ messageRecord.setSize(message.getSize());
+ messageRecord.setValue(new Buffer(mos.toBuffer().getData()));
+ return messageRecord;
+ }
+
+ public File getFile() {
+ return file;
+ }
+
+ public void setFilename(String filename) {
+ this.filename = filename;
+ }
+
+ public String getFilename() {
+ return filename;
+ }
+
+ public void setFile(File file) {
+ this.file = file;
+ }
+
+ public URI getConfig() {
+ return config;
+ }
+
+ public void setConfig(URI config) {
+ this.config = config;
+ }
+
+ public void setQueue(String queue) {
+ this.queue = queue;
+ }
+
+ public String getQueue() {
+ return queue;
+ }
+
+ public void setOffset(int offset) {
+ this.offset = offset;
+ }
+
+ public Integer getOffset() {
+ return offset;
+ }
+
+ public void setCount(int count) {
+ this.count = count;
+ }
+
+ public Integer getCount() {
+ return count;
+ }
+}
diff --git a/activemq-console/src/main/protobuf/data.proto b/activemq-console/src/main/protobuf/data.proto
new file mode 100644
index 0000000000..0cc765d079
--- /dev/null
+++ b/activemq-console/src/main/protobuf/data.proto
@@ -0,0 +1,60 @@
+//
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+package org.apache.activemq.console.command.store.protobuf;
+
+option java_multiple_files = true;
+
+///////////////////////////////////////////////////////////////
+// Message related operations.
+///////////////////////////////////////////////////////////////
+
+message MessagePB {
+ required int64 messageKey=1;
+ required bytes codec = 2 [java_override_type = "AsciiBuffer"];
+ optional int32 size = 3;
+ optional bytes value = 4;
+ optional sint64 expiration = 5;
+ optional int32 compression = 6;
+
+ optional bytes direct_data = 10;
+ optional bytes direct_file = 12;
+ optional int64 direct_offset = 13;
+ optional int32 direct_size = 14;
+}
+
+message QueuePB {
+ required int64 key=1;
+ optional bytes binding_kind = 2 [java_override_type = "AsciiBuffer"];
+ optional bytes binding_data = 3;
+}
+
+message QueueEntryPB {
+ required int64 queueKey=1;
+ required int64 queueSeq=2;
+ required int64 messageKey=3;
+ optional int32 size=4;
+ optional bytes attachment=5;
+ optional int32 redeliveries = 6;
+ optional sint64 expiration=7;
+ optional bytes messageLocator=8;
+ repeated bytes sender=9;
+}
+
+message MapEntryPB {
+ required bytes key = 1;
+ optional bytes value = 2;
+}
diff --git a/activemq-console/src/main/resources/META-INF/services/org.apache.activemq.console.command.Command b/activemq-console/src/main/resources/META-INF/services/org.apache.activemq.console.command.Command
index bb7ddb629c..7df27efab8 100644
--- a/activemq-console/src/main/resources/META-INF/services/org.apache.activemq.console.command.Command
+++ b/activemq-console/src/main/resources/META-INF/services/org.apache.activemq.console.command.Command
@@ -28,3 +28,4 @@ org.apache.activemq.console.command.StoreExportCommand
org.apache.activemq.console.command.PurgeCommand
org.apache.activemq.console.command.ProducerCommand
org.apache.activemq.console.command.ConsumerCommand
+org.apache.activemq.console.command.StoreBackupCommand
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@activemq.apache.org
For additional commands, e-mail: commits-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact