001/**
002 * Licensed to the Apache Software Foundation (ASF) under one or more
003 * contributor license agreements.  See the NOTICE file distributed with
004 * this work for additional information regarding copyright ownership.
005 * The ASF licenses this file to You under the Apache License, Version 2.0
006 * (the "License"); you may not use this file except in compliance with
007 * the License.  You may obtain a copy of the License at
008 *
009 *      http://www.apache.org/licenses/LICENSE-2.0
010 *
011 * Unless required by applicable law or agreed to in writing, software
012 * distributed under the License is distributed on an "AS IS" BASIS,
013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
014 * See the License for the specific language governing permissions and
015 * limitations under the License.
016 */
017package org.apache.activemq.store.kahadb;
018
019import java.io.File;
020import java.io.IOException;
021import java.util.Date;
022import java.util.Map;
023import java.util.Set;
024import java.util.TreeSet;
025import java.util.concurrent.ConcurrentHashMap;
026import java.util.concurrent.ConcurrentMap;
027import java.util.concurrent.atomic.AtomicBoolean;
028
029import org.apache.activemq.broker.Broker;
030import org.apache.activemq.broker.ConnectionContext;
031import org.apache.activemq.broker.region.BaseDestination;
032import org.apache.activemq.command.Message;
033import org.apache.activemq.command.MessageAck;
034import org.apache.activemq.command.MessageId;
035import org.apache.activemq.command.TransactionId;
036import org.apache.activemq.command.XATransactionId;
037import org.apache.activemq.store.AbstractMessageStore;
038import org.apache.activemq.store.IndexListener;
039import org.apache.activemq.store.ListenableFuture;
040import org.apache.activemq.store.MessageStore;
041import org.apache.activemq.store.PersistenceAdapter;
042import org.apache.activemq.store.ProxyMessageStore;
043import org.apache.activemq.store.ProxyTopicMessageStore;
044import org.apache.activemq.store.TopicMessageStore;
045import org.apache.activemq.store.TransactionRecoveryListener;
046import org.apache.activemq.store.TransactionStore;
047import org.apache.activemq.store.kahadb.data.KahaCommitCommand;
048import org.apache.activemq.store.kahadb.data.KahaEntryType;
049import org.apache.activemq.store.kahadb.data.KahaPrepareCommand;
050import org.apache.activemq.store.kahadb.data.KahaTraceCommand;
051import org.apache.activemq.store.kahadb.disk.journal.DataFile;
052import org.apache.activemq.store.kahadb.disk.journal.Journal;
053import org.apache.activemq.store.kahadb.disk.journal.Location;
054import org.apache.activemq.usage.StoreUsage;
055import org.apache.activemq.util.DataByteArrayInputStream;
056import org.apache.activemq.util.DataByteArrayOutputStream;
057import org.apache.activemq.util.IOHelper;
058import org.slf4j.Logger;
059import org.slf4j.LoggerFactory;
060
061public class MultiKahaDBTransactionStore implements TransactionStore {
062    static final Logger LOG = LoggerFactory.getLogger(MultiKahaDBTransactionStore.class);
063    final MultiKahaDBPersistenceAdapter multiKahaDBPersistenceAdapter;
064    final ConcurrentMap<TransactionId, Tx> inflightTransactions = new ConcurrentHashMap<TransactionId, Tx>();
065    final ConcurrentMap<TransactionId, Tx> pendingCommit = new ConcurrentHashMap<TransactionId, Tx>();
066    private Journal journal;
067    private int journalMaxFileLength = Journal.DEFAULT_MAX_FILE_LENGTH;
068    private int journalWriteBatchSize = Journal.DEFAULT_MAX_WRITE_BATCH_SIZE;
069    private final AtomicBoolean started = new AtomicBoolean(false);
070    private final AtomicBoolean recovered = new AtomicBoolean(false);
071    private long journalCleanupInterval = Journal.DEFAULT_CLEANUP_INTERVAL;
072    private boolean checkForCorruption = true;
073    private AtomicBoolean corruptJournalDetected = new AtomicBoolean(false);
074
075    public MultiKahaDBTransactionStore(MultiKahaDBPersistenceAdapter multiKahaDBPersistenceAdapter) {
076        this.multiKahaDBPersistenceAdapter = multiKahaDBPersistenceAdapter;
077    }
078
079    public MessageStore proxy(final TransactionStore transactionStore, MessageStore messageStore) {
080        return new ProxyMessageStore(messageStore) {
081            @Override
082            public void addMessage(ConnectionContext context, final Message send) throws IOException {
083                MultiKahaDBTransactionStore.this.addMessage(transactionStore, context, getDelegate(), send);
084            }
085
086            @Override
087            public void addMessage(ConnectionContext context, final Message send, boolean canOptimizeHint) throws IOException {
088                MultiKahaDBTransactionStore.this.addMessage(transactionStore, context, getDelegate(), send);
089            }
090
091            @Override
092            public ListenableFuture<Object> asyncAddQueueMessage(ConnectionContext context, Message message) throws IOException {
093                return MultiKahaDBTransactionStore.this.asyncAddQueueMessage(transactionStore, context, getDelegate(), message);
094            }
095
096            @Override
097            public ListenableFuture<Object> asyncAddQueueMessage(ConnectionContext context, Message message, boolean canOptimizeHint) throws IOException {
098                return MultiKahaDBTransactionStore.this.asyncAddQueueMessage(transactionStore, context, getDelegate(), message);
099            }
100
101            @Override
102            public void removeMessage(ConnectionContext context, final MessageAck ack) throws IOException {
103                MultiKahaDBTransactionStore.this.removeMessage(transactionStore, context, getDelegate(), ack);
104            }
105
106            @Override
107            public void removeAsyncMessage(ConnectionContext context, MessageAck ack) throws IOException {
108                MultiKahaDBTransactionStore.this.removeAsyncMessage(transactionStore, context, getDelegate(), ack);
109            }
110
111            @Override
112            public void registerIndexListener(IndexListener indexListener) {
113                getDelegate().registerIndexListener(indexListener);
114                try {
115                    if (indexListener instanceof BaseDestination) {
116                        // update queue storeUsage
117                        Object matchingPersistenceAdapter = multiKahaDBPersistenceAdapter.destinationMap.chooseValue(getDelegate().getDestination());
118                        if (matchingPersistenceAdapter instanceof FilteredKahaDBPersistenceAdapter) {
119                            FilteredKahaDBPersistenceAdapter filteredAdapter = (FilteredKahaDBPersistenceAdapter) matchingPersistenceAdapter;
120                            if (filteredAdapter.getUsage() != null && filteredAdapter.getPersistenceAdapter() instanceof KahaDBPersistenceAdapter) {
121                                StoreUsage storeUsage = filteredAdapter.getUsage();
122                                storeUsage.setStore(filteredAdapter.getPersistenceAdapter());
123                                storeUsage.setParent(multiKahaDBPersistenceAdapter.getBrokerService().getSystemUsage().getStoreUsage());
124                                ((BaseDestination) indexListener).getSystemUsage().setStoreUsage(storeUsage);
125                            }
126                        }
127                    }
128                } catch (Exception ignored) {
129                    LOG.warn("Failed to set mKahaDB destination store usage", ignored);
130                }
131            }
132        };
133    }
134
135    public TopicMessageStore proxy(final TransactionStore transactionStore, final TopicMessageStore messageStore) {
136        return new ProxyTopicMessageStore(messageStore) {
137            @Override
138            public void addMessage(ConnectionContext context, final Message send, boolean canOptimizeHint) throws IOException {
139                MultiKahaDBTransactionStore.this.addMessage(transactionStore, context, getDelegate(), send);
140            }
141
142            @Override
143            public void addMessage(ConnectionContext context, final Message send) throws IOException {
144                MultiKahaDBTransactionStore.this.addMessage(transactionStore, context, getDelegate(), send);
145            }
146
147            @Override
148            public ListenableFuture<Object> asyncAddTopicMessage(ConnectionContext context, Message message, boolean canOptimizeHint) throws IOException {
149                return MultiKahaDBTransactionStore.this.asyncAddTopicMessage(transactionStore, context, getDelegate(), message);
150            }
151
152            @Override
153            public ListenableFuture<Object> asyncAddTopicMessage(ConnectionContext context, Message message) throws IOException {
154                return MultiKahaDBTransactionStore.this.asyncAddTopicMessage(transactionStore, context, getDelegate(), message);
155            }
156
157            @Override
158            public void removeMessage(ConnectionContext context, final MessageAck ack) throws IOException {
159                MultiKahaDBTransactionStore.this.removeMessage(transactionStore, context, getDelegate(), ack);
160            }
161
162            @Override
163            public void removeAsyncMessage(ConnectionContext context, MessageAck ack) throws IOException {
164                MultiKahaDBTransactionStore.this.removeAsyncMessage(transactionStore, context, getDelegate(), ack);
165            }
166
167            @Override
168            public void acknowledge(ConnectionContext context, String clientId, String subscriptionName,
169                                    MessageId messageId, MessageAck ack) throws IOException {
170                MultiKahaDBTransactionStore.this.acknowledge(transactionStore, context, (TopicMessageStore) getDelegate(), clientId,
171                        subscriptionName, messageId, ack);
172            }
173        };
174    }
175
176    public void deleteAllMessages() {
177        IOHelper.deleteChildren(getDirectory());
178    }
179
180    public int getJournalMaxFileLength() {
181        return journalMaxFileLength;
182    }
183
184    public void setJournalMaxFileLength(int journalMaxFileLength) {
185        this.journalMaxFileLength = journalMaxFileLength;
186    }
187
188    public int getJournalMaxWriteBatchSize() {
189        return journalWriteBatchSize;
190    }
191
192    public void setJournalMaxWriteBatchSize(int journalWriteBatchSize) {
193        this.journalWriteBatchSize = journalWriteBatchSize;
194    }
195
196    public void setJournalCleanupInterval(long journalCleanupInterval) {
197        this.journalCleanupInterval = journalCleanupInterval;
198    }
199
200    public long getJournalCleanupInterval() {
201        return journalCleanupInterval;
202    }
203
204    public void setCheckForCorruption(boolean checkForCorruption) {
205        this.checkForCorruption = checkForCorruption;
206    }
207
208    public boolean isCheckForCorruption() {
209        return checkForCorruption;
210    }
211
212    private static final XATransactionId NULL_XA_TRANSACTION_ID = new XATransactionId();
213    public class Tx {
214        private final ConcurrentHashMap<TransactionStore, TransactionId> stores = new ConcurrentHashMap<TransactionStore, TransactionId>();
215        private int prepareLocationId = 0;
216
217        public void trackStore(TransactionStore store, XATransactionId xid) {
218            stores.put(store, xid);
219        }
220
221        public void trackStore(TransactionStore store) {
222            stores.putIfAbsent(store, NULL_XA_TRANSACTION_ID);
223        }
224
225        public Map<TransactionStore, TransactionId> getStoresMap() {
226            return stores;
227        }
228
229        public Set<TransactionStore> getStores() {
230            return stores.keySet();
231        }
232
233        public void trackPrepareLocation(Location location) {
234            this.prepareLocationId = location.getDataFileId();
235        }
236
237        public int getPreparedLocationId() {
238            return prepareLocationId;
239        }
240    }
241
242    public Tx getTx(TransactionId txid) {
243        Tx tx = inflightTransactions.get(txid);
244        if (tx == null) {
245            final Tx val = new Tx();
246            tx = inflightTransactions.putIfAbsent(txid, val);
247            if (tx == null) {
248                // we won
249                tx = val;
250            }
251        }
252        return tx;
253    }
254
255    public Tx removeTx(TransactionId txid) {
256        return inflightTransactions.remove(txid);
257    }
258
259    @Override
260    public void prepare(TransactionId txid) throws IOException {
261        Tx tx = getTx(txid);
262        for (TransactionStore store : tx.getStores()) {
263            store.prepare(txid);
264        }
265    }
266
267    @Override
268    public void commit(TransactionId txid, boolean wasPrepared, Runnable preCommit, Runnable postCommit)
269            throws IOException {
270
271        if (preCommit != null) {
272            preCommit.run();
273        }
274
275        Tx tx = getTx(txid);
276        if (wasPrepared) {
277            for (Map.Entry<TransactionStore, TransactionId> storeTx : tx.getStoresMap().entrySet()) {
278                TransactionId recovered = storeTx.getValue();
279                if (recovered != null && recovered != NULL_XA_TRANSACTION_ID ) {
280                    storeTx.getKey().commit(recovered, true, null, null);
281                } else {
282                    storeTx.getKey().commit(txid, true, null, null);
283                }
284            }
285        } else {
286            // can only do 1pc on a single store
287            if (tx.getStores().size() == 1) {
288                for (TransactionStore store : tx.getStores()) {
289                    store.commit(txid, false, null, null);
290                }
291            } else {
292                // need to do local 2pc
293                for (TransactionStore store : tx.getStores()) {
294                    store.prepare(txid);
295                }
296                persistOutcome(tx, txid);
297                for (TransactionStore store : tx.getStores()) {
298                    store.commit(txid, true, null, null);
299                }
300                persistCompletion(txid);
301            }
302        }
303        removeTx(txid);
304        if (postCommit != null) {
305            postCommit.run();
306        }
307    }
308
309    public void persistOutcome(Tx tx, TransactionId txid) throws IOException {
310        tx.trackPrepareLocation(store(new KahaPrepareCommand().setTransactionInfo(TransactionIdConversion.convert(multiKahaDBPersistenceAdapter.transactionIdTransformer.transform(txid)))));
311        pendingCommit.put(txid, tx);
312    }
313
314    public void persistCompletion(TransactionId txid) throws IOException {
315        store(new KahaCommitCommand().setTransactionInfo(TransactionIdConversion.convert(multiKahaDBPersistenceAdapter.transactionIdTransformer.transform(txid))));
316        pendingCommit.remove(txid);
317    }
318
319    private Location store(JournalCommand<?> data) throws IOException {
320        int size = data.serializedSizeFramed();
321        DataByteArrayOutputStream os = new DataByteArrayOutputStream(size + 1);
322        os.writeByte(data.type().getNumber());
323        data.writeFramed(os);
324        Location location = journal.write(os.toByteSequence(), true);
325        journal.setLastAppendLocation(location);
326        return location;
327    }
328
329    @Override
330    public void rollback(TransactionId txid) throws IOException {
331        Tx tx = removeTx(txid);
332        if (tx != null) {
333            for (Map.Entry<TransactionStore, TransactionId> storeTx : tx.getStoresMap().entrySet()) {
334                TransactionId recovered = storeTx.getValue();
335                if (recovered != null && recovered != NULL_XA_TRANSACTION_ID) {
336                    storeTx.getKey().rollback(recovered);
337                } else {
338                    storeTx.getKey().rollback(txid);
339                }
340            }
341        }
342    }
343
344    @Override
345    public void start() throws Exception {
346        if (started.compareAndSet(false, true)) {
347            journal = new Journal() {
348                @Override
349                public void cleanup() {
350                    super.cleanup();
351                    txStoreCleanup();
352                }
353            };
354            journal.setDirectory(getDirectory());
355            journal.setMaxFileLength(journalMaxFileLength);
356            journal.setWriteBatchSize(journalWriteBatchSize);
357            journal.setCleanupInterval(journalCleanupInterval);
358            journal.setCheckForCorruptionOnStartup(checkForCorruption);
359            journal.setChecksum(checkForCorruption);
360            IOHelper.mkdirs(journal.getDirectory());
361            journal.start();
362            recoverPendingLocalTransactions();
363            recovered.set(true);
364            loaded();
365        }
366    }
367
368    private void loaded() throws IOException {
369        store(new KahaTraceCommand().setMessage("LOADED " + new Date()));
370    }
371
372    private void txStoreCleanup() {
373        if (!recovered.get() || corruptJournalDetected.get()) {
374            return;
375        }
376        Set<Integer> knownDataFileIds = new TreeSet<Integer>(journal.getFileMap().keySet());
377        for (Tx tx : inflightTransactions.values()) {
378            knownDataFileIds.remove(tx.getPreparedLocationId());
379        }
380        for (Tx tx : pendingCommit.values()) {
381            knownDataFileIds.remove(tx.getPreparedLocationId());
382        }
383        try {
384            journal.removeDataFiles(knownDataFileIds);
385        } catch (Exception e) {
386            LOG.error(this + ", Failed to remove tx journal datafiles " + knownDataFileIds);
387        }
388    }
389
390    private File getDirectory() {
391        return new File(multiKahaDBPersistenceAdapter.getDirectory(), "txStore");
392    }
393
394    @Override
395    public void stop() throws Exception {
396        if (started.compareAndSet(true, false) && journal != null) {
397            journal.close();
398            journal = null;
399        }
400    }
401
402    private void recoverPendingLocalTransactions() throws IOException {
403
404        if (checkForCorruption) {
405            for (DataFile dataFile: journal.getFileMap().values()) {
406                if (!dataFile.getCorruptedBlocks().isEmpty()) {
407                    LOG.error("Corrupt Transaction journal records found in db-{}.log at {}", dataFile.getDataFileId(),  dataFile.getCorruptedBlocks());
408                    corruptJournalDetected.set(true);
409                }
410            }
411        }
412        if (!corruptJournalDetected.get()) {
413            Location location = null;
414            try {
415                location = journal.getNextLocation(null);
416                while (location != null) {
417                    process(location, load(location));
418                    location = journal.getNextLocation(location);
419                }
420            } catch (Exception oops) {
421                LOG.error("Corrupt journal record; unexpected exception on transaction journal replay of location:" + location, oops);
422                corruptJournalDetected.set(true);
423            }
424            pendingCommit.putAll(inflightTransactions);
425            LOG.info("pending local transactions: " + pendingCommit.keySet());
426        }
427    }
428
429    public JournalCommand<?> load(Location location) throws IOException {
430        DataByteArrayInputStream is = new DataByteArrayInputStream(journal.read(location));
431        byte readByte = is.readByte();
432        KahaEntryType type = KahaEntryType.valueOf(readByte);
433        if (type == null) {
434            throw new IOException("Could not load journal record. Invalid location: " + location);
435        }
436        JournalCommand<?> message = (JournalCommand<?>) type.createMessage();
437        message.mergeFramed(is);
438        return message;
439    }
440
441    public void process(final Location location, JournalCommand<?> command) throws IOException {
442        switch (command.type()) {
443            case KAHA_PREPARE_COMMAND:
444                KahaPrepareCommand prepareCommand = (KahaPrepareCommand) command;
445                getTx(TransactionIdConversion.convert(prepareCommand.getTransactionInfo())).trackPrepareLocation(location);
446                break;
447            case KAHA_COMMIT_COMMAND:
448                KahaCommitCommand commitCommand = (KahaCommitCommand) command;
449                removeTx(TransactionIdConversion.convert(commitCommand.getTransactionInfo()));
450                break;
451            case KAHA_TRACE_COMMAND:
452                break;
453            default:
454                throw new IOException("Unexpected command in transaction journal: " + command);
455        }
456    }
457
458
459    @Override
460    public synchronized void recover(final TransactionRecoveryListener listener) throws IOException {
461
462        for (final PersistenceAdapter adapter : multiKahaDBPersistenceAdapter.adapters) {
463            adapter.createTransactionStore().recover(new TransactionRecoveryListener() {
464                @Override
465                public void recover(XATransactionId xid, Message[] addedMessages, MessageAck[] acks) {
466                    try {
467                        getTx(xid).trackStore(adapter.createTransactionStore(), xid);
468                    } catch (IOException e) {
469                        LOG.error("Failed to access transaction store: " + adapter + " for prepared xa tid: " + xid, e);
470                    }
471                    listener.recover(xid, addedMessages, acks);
472                }
473            });
474        }
475
476        boolean recoveryWorkPending = false;
477        try {
478            Broker broker = multiKahaDBPersistenceAdapter.getBrokerService().getBroker();
479            // force completion of local xa
480            for (TransactionId txid : broker.getPreparedTransactions(null)) {
481                if (multiKahaDBPersistenceAdapter.isLocalXid(txid)) {
482                    recoveryWorkPending = true;
483                    if (corruptJournalDetected.get()) {
484                        // not having a record is meaningless once our tx store is corrupt; we need a heuristic decision
485                        LOG.warn("Pending multi store local transaction {} requires manual heuristic outcome via JMX", txid);
486                        logSomeContext(txid);
487                    } else {
488                        try {
489                            if (pendingCommit.keySet().contains(txid)) {
490                                // we recorded the commit outcome, finish the job
491                                LOG.info("delivering pending commit outcome for tid: " + txid);
492                                broker.commitTransaction(null, txid, false);
493                            } else {
494                                // we have not record an outcome, and would have reported a commit failure, so we must rollback
495                                LOG.info("delivering rollback outcome to store for tid: " + txid);
496                                broker.forgetTransaction(null, txid);
497                            }
498                            persistCompletion(txid);
499                        } catch (Exception ex) {
500                            LOG.error("failed to deliver pending outcome for tid: " + txid, ex);
501                        }
502                    }
503                }
504            }
505        } catch (Exception e) {
506            LOG.error("failed to resolve pending local transactions", e);
507        }
508        // can we ignore corruption and resume
509        if (corruptJournalDetected.get() && !recoveryWorkPending) {
510            // move to new write file, gc will cleanup
511            journal.rotateWriteFile();
512            loaded();
513            corruptJournalDetected.set(false);
514            LOG.info("No heuristics outcome pending after corrupt tx store detection, auto resolving");
515        }
516    }
517
518    private void logSomeContext(TransactionId txid) throws IOException {
519        Tx tx = getTx(txid);
520        if (tx != null) {
521            for (TransactionStore store: tx.getStores()) {
522                for (PersistenceAdapter persistenceAdapter : multiKahaDBPersistenceAdapter.adapters) {
523                    if (persistenceAdapter.createTransactionStore() == store) {
524                        if (persistenceAdapter instanceof KahaDBPersistenceAdapter) {
525                            LOG.warn("Heuristic data in: " + persistenceAdapter + ", "  + ((KahaDBPersistenceAdapter)persistenceAdapter).getStore().getPreparedTransaction(txid));
526                        }
527                    }
528                }
529            }
530        }
531    }
532
533    void addMessage(final TransactionStore transactionStore, ConnectionContext context, final MessageStore destination, final Message message)
534            throws IOException {
535        if (message.getTransactionId() != null) {
536            getTx(message.getTransactionId()).trackStore(transactionStore);
537        }
538        destination.addMessage(context, message);
539    }
540
541    ListenableFuture<Object> asyncAddQueueMessage(final TransactionStore transactionStore, ConnectionContext context, final MessageStore destination, final Message message)
542            throws IOException {
543        if (message.getTransactionId() != null) {
544            getTx(message.getTransactionId()).trackStore(transactionStore);
545            destination.addMessage(context, message);
546            return AbstractMessageStore.FUTURE;
547        } else {
548            return destination.asyncAddQueueMessage(context, message);
549        }
550    }
551
552    ListenableFuture<Object> asyncAddTopicMessage(final TransactionStore transactionStore, ConnectionContext context, final MessageStore destination, final Message message)
553            throws IOException {
554
555        if (message.getTransactionId() != null) {
556            getTx(message.getTransactionId()).trackStore(transactionStore);
557            destination.addMessage(context, message);
558            return AbstractMessageStore.FUTURE;
559        } else {
560            return destination.asyncAddTopicMessage(context, message);
561        }
562    }
563
564    final void removeMessage(final TransactionStore transactionStore, ConnectionContext context, final MessageStore destination, final MessageAck ack)
565            throws IOException {
566        if (ack.getTransactionId() != null) {
567            getTx(ack.getTransactionId()).trackStore(transactionStore);
568        }
569        destination.removeMessage(context, ack);
570    }
571
572    final void removeAsyncMessage(final TransactionStore transactionStore, ConnectionContext context, final MessageStore destination, final MessageAck ack)
573            throws IOException {
574        if (ack.getTransactionId() != null) {
575            getTx(ack.getTransactionId()).trackStore(transactionStore);
576        }
577        destination.removeAsyncMessage(context, ack);
578    }
579
580    final void acknowledge(final TransactionStore transactionStore, ConnectionContext context, final TopicMessageStore destination,
581                           final String clientId, final String subscriptionName,
582                           final MessageId messageId, final MessageAck ack) throws IOException {
583        if (ack.getTransactionId() != null) {
584            getTx(ack.getTransactionId()).trackStore(transactionStore);
585        }
586        destination.acknowledge(context, clientId, subscriptionName, messageId, ack);
587    }
588
589}