Confluence Journal Service
Confluence 5.6 and later
Confluence Journal Service provides access to journals that are durable FIFO queues with the following behavior:
- Journal entries are processed on all nodes when running in a cluster.
- Entries are durable - receiving nodes do not need to be running when the event is queued.
- New cluster nodes can process entries that were queued before the node was created.
- Entries are guaranteed to be delivered in the same order that they were added.
- Journals are bounded in size by the age of the entries, i.e. entries that are older than a set limit are cleared regularly.
Journals are accessed through
Each journal has its own journal identifier. Journals do not need to be explicitly created or removed by client code.
Creating journal entries
Journal entries have five fields:
id: Monotonically increasing unique identifier for the entry. It is assigned automatically when an entry is queued. An entry with a larger id is considered to have happened after entries with a smaller id.
journalId: Identifies the journal that the entry belongs to. Maximum length is 255 characters. May only contain lower-case letters (English alphabet), digits, underscores and dot characters.
creationDate: Point in time when the entry was created. It is assigned automatically when an entry is queued.
type: Type of the entry. This is a journal specific free-form field. Maximum length is 255 characters.
message: Message of the entry. This is a journal specific free-form field. Maximum length is 2047 characters.
Processing journal entries
Entries are processed by calling
JournalService#processEntries method. It has the following signature:
processEntries will call the given
entryProcessor function with at most
maxEntries entries which have the given
entryProcessor will in turn create an instance of
EntryProcessorResult that contains a
result value that is returned to the caller of
processEntries. Depending on
entryProcessor return value, four things can happen on next call to
EntryProcessorResult.success(result)causes none of the processed entries to be processed again.
EntryProcessorResult.partial(result, lastSuccessfulId)causes entries with an id larger than
lastSuccessfulIdto be processed again.
EntryProcessorResult.failure(result, failedEntryId)causes entries with an id equal to or larger than failedEntryId to be processed again.
- Throwing a
RuntimeExceptioncauses all the entries to be processed again.
Entries should be processed in small batches (i.e. small
maxEntries value) in order to reduce memory consumption during the processing. The following example shows how to process all the entries in batches. It also demonstrates how to ensure that already processed entries will not be processed again if a
RuntimeException is thrown when processing an entry.
JournalService also offers methods for retrieving current entries in the journal without removing them from journal (
peek), removing entries from journal without processing them (
reset) and counting the number of entries in the journal (
Usage in cluster
Each cluster node keeps track of the entries that have been processed on the same node. This means each entry is processed on all cluster nodes.
A compatibility library is available for plugins that want to use Journal Service but need to stay compatible with older Confluence versions. In older versions of Confluence (5.5 and earlier) entries are stored in main memory. This means that unprocessed entries will be forgotten during restart and that entries are only processed on the local node in a cluster.
In addition to maven dependency, the compatibility library depends on Journal Service on Confluence 5.6 and later, thus related packaged need to be dynamically imported in OSGi instructions section in the plugin pom.xml:
com.atlassian.confluence.journal.compat.JournalServiceFactory is the entry point to the compatibility library, and it can be used to get a JournalService instance:
Normally tests would invoke
JournalService#waitForRecentEntriesToBecomeVisible() after running code that added entries and before running code that needs to process those entries. If there are a lot of tests like this, the wait time can become significant. The following example code shows how to use the
com.atlassian.confluence.test.JournalManagerBackdoor service for temporarily disabling this wait. It should only be used in test code.