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
The following commit(s) were added to refs/heads/main by this push:
new a148e598ab [#2140] Improve KahaDB extensibility
a148e598ab is described below
commit a148e598ab065656523f87e54f77900cce44dd07
Author: Matt Pavlovich <[email protected]>
AuthorDate: Mon Jun 22 21:56:34 2026 -0500
[#2140] Improve KahaDB extensibility
---
.../store/kahadb/KahaDBPersistenceAdapter.java | 6 +++++-
.../apache/activemq/store/kahadb/MessageDatabase.java | 18 +++++++++---------
.../activemq/store/kahadb/disk/journal/DataFile.java | 2 +-
3 files changed, 15 insertions(+), 11 deletions(-)
diff --git
a/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/KahaDBPersistenceAdapter.java
b/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/KahaDBPersistenceAdapter.java
index 1955f8450b..2725eb08a9 100644
---
a/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/KahaDBPersistenceAdapter.java
+++
b/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/KahaDBPersistenceAdapter.java
@@ -67,7 +67,11 @@ import static
org.apache.activemq.broker.jmx.BrokerMBeanSupport.createPersistenc
public class KahaDBPersistenceAdapter extends LockableServiceSupport
implements PersistenceAdapter,
JournaledStore, TransactionIdTransformerAware, NoLocalSubscriptionAware {
- private final KahaDBStore letter = new KahaDBStore();
+ private final KahaDBStore letter = createStore();
+
+ protected KahaDBStore createStore() {
+ return new KahaDBStore();
+ }
/**
* @param context
diff --git
a/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java
b/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java
index 2709817d01..08751b9c3b 100644
---
a/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java
+++
b/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java
@@ -138,7 +138,7 @@ public abstract class MessageDatabase extends
ServiceSupport implements BrokerSe
static final byte COMPACTED_JOURNAL_FILE = DataFile.STANDARD_LOG_FILE + 1;
- class Metadata {
+ protected class Metadata {
protected Page<Metadata> page;
protected int state;
protected BTreeIndex<String, StoredDestination> destinations;
@@ -249,7 +249,7 @@ public abstract class MessageDatabase extends
ServiceSupport implements BrokerSe
protected PageFile pageFile;
protected Journal journal;
- Metadata metadata = new Metadata();
+ protected Metadata metadata = new Metadata();
protected final PersistenceAdapterStatistics persistenceAdapterStatistics
= new PersistenceAdapterStatistics();
MetadataMarshaller metadataMarshaller = new MetadataMarshaller();
@@ -1258,7 +1258,7 @@ public abstract class MessageDatabase extends
ServiceSupport implements BrokerSe
}
@SuppressWarnings("rawtypes")
- void process(final KahaAddMessageCommand command, final Location location,
final IndexAware runWithIndexLock) throws IOException {
+ protected void process(final KahaAddMessageCommand command, final Location
location, final IndexAware runWithIndexLock) throws IOException {
if (command.hasTransactionInfo()) {
List<Operation> inflightTx =
getInflightTx(command.getTransactionInfo());
inflightTx.add(new AddOperation(command, location,
runWithIndexLock));
@@ -1334,7 +1334,7 @@ public abstract class MessageDatabase extends
ServiceSupport implements BrokerSe
}
@SuppressWarnings("rawtypes")
- void process(KahaCommitCommand command, final Location location, final
IndexAware before) throws IOException {
+ protected void process(KahaCommitCommand command, final Location location,
final IndexAware before) throws IOException {
TransactionId key =
TransactionIdConversion.convert(command.getTransactionInfo());
List<Operation> inflightTx;
synchronized (inflightTransactions) {
@@ -3179,7 +3179,7 @@ public abstract class MessageDatabase extends
ServiceSupport implements BrokerSe
@SuppressWarnings("rawtypes")
private final LinkedHashMap<TransactionId, List<Operation>>
inflightTransactions = new LinkedHashMap<>();
@SuppressWarnings("rawtypes")
- final LinkedHashMap<TransactionId, List<Operation>> preparedTransactions =
new LinkedHashMap<>();
+ protected final LinkedHashMap<TransactionId, List<Operation>>
preparedTransactions = new LinkedHashMap<>();
@SuppressWarnings("rawtypes")
private List<Operation> getInflightTx(KahaTransactionInfo info) {
@@ -3197,7 +3197,7 @@ public abstract class MessageDatabase extends
ServiceSupport implements BrokerSe
return TransactionIdConversion.convert(transactionInfo);
}
- abstract static class Operation <T extends JournalCommand<T>> {
+ protected abstract static class Operation <T extends JournalCommand<T>> {
final T command;
final Location location;
@@ -3217,7 +3217,7 @@ public abstract class MessageDatabase extends
ServiceSupport implements BrokerSe
abstract public void execute(Transaction tx) throws IOException;
}
- class AddOperation extends Operation<KahaAddMessageCommand> {
+ protected class AddOperation extends Operation<KahaAddMessageCommand> {
final IndexAware runWithIndexLock;
public AddOperation(KahaAddMessageCommand command, Location location,
IndexAware runWithIndexLock) {
super(command, location);
@@ -3233,7 +3233,7 @@ public abstract class MessageDatabase extends
ServiceSupport implements BrokerSe
}
}
- class RemoveOperation extends Operation<KahaRemoveMessageCommand> {
+ protected class RemoveOperation extends
Operation<KahaRemoveMessageCommand> {
public RemoveOperation(KahaRemoveMessageCommand command, Location
location) {
super(command, location);
@@ -4024,7 +4024,7 @@ public abstract class MessageDatabase extends
ServiceSupport implements BrokerSe
this.indexDirectory = indexDirectory;
}
- interface IndexAware {
+ protected interface IndexAware {
void sequenceAssignedWithIndexLocked(long index);
}
diff --git
a/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/disk/journal/DataFile.java
b/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/disk/journal/DataFile.java
index 5e8fb7104e..fc4edeb7bc 100644
---
a/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/disk/journal/DataFile.java
+++
b/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/disk/journal/DataFile.java
@@ -38,7 +38,7 @@ public class DataFile extends LinkedNode<DataFile> implements
Comparable<DataFil
protected final SequenceSet corruptedBlocks = new SequenceSet();
protected RecoverableRandomAccessFile appendRandomAccessFile;
- DataFile(File file, int number) {
+ protected DataFile(File file, int number) {
this.file = file;
this.dataFileId = number;
length = (int)(file.exists() ? file.length() : 0);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
For further information, visit: https://activemq.apache.org/contact
