com.thinkaurelius.titan.diskstorage.log.kcvs

## Class KCVSLog

- java.lang.Object
- - com.thinkaurelius.titan.diskstorage.log.kcvs.KCVSLog

- All Implemented Interfaces:
    Log, BackendOperation.TransactionalProvider

      public class KCVSLog
      extends Object
      implements Log, BackendOperation.TransactionalProvider

  Implementation of `Log` wrapped around a `KeyColumnValueStore`. Each message is written as a column-value pair (`Entry`) into a timeslice slot. A timeslice slot is uniquely identified by:
  - The partition id: On storage backends that are key-ordered, a partition bit width can be configured which configures the number of first bits that comprise the partition id. On unordered storage backends, this is always 0
  - A bucket id: The number of parallel buckets that should be maintained is configured by `GraphDatabaseConfiguration.LOG_NUM_BUCKETS`. Messages are written to the buckets in round-robin fashion and each bucket is identified by a bucket id. Having multiple buckets per timeslice allows for load balancing across multiple keys in the storage backend.
  - The start time of the timeslice: Each time slice is `TIMESLICE_INTERVAL` microseconds long. And all messages that are added between start-time and start-time+`TIMESLICE_INTERVAL` end up in the same timeslice. For high throughput logs that might be more messages than the underlying storage backend can handle per key. In that case, ensure that (2^(partition-bit-width) x (num-bucekts) is large enough to distribute the load.
  Each message is uniquely identified by its timestamp, sender id (which uniquely identifies a particular instance of `KCVSLogManager`), and the message id (which is auto-incrementing). These three data points comprise the column of a log message. The actual content of the message is written into the value. When `MessageReader` are registered, one reader thread per partition id and bucket is created which periodically (as configured) checks for new messages in the storage backend and invokes the reader. Read-markers are maintained (for each partition-id & bucket id combination) under a dedicated key in the same `KeyColumnValueStoreManager` as the log messages. The read markers are updated to the current position before each new iteration of reading messages from the log. If the system fails while reading a batch of messages, a subsequently restarted log reader may therefore read messages twice. Hence, `MessageReader` implementations should exhibit correct behavior for the (rare) circumstance that messages are read twice. Note: All time values in this class are in mircoseconds. Hence, there are many cases where milliseconds are converted to microseconds.
  Author:
    Matthias Broecheler (me@matthiasb.com)

- - ### Field Summary

    Fields
    Modifier and Type | Field and Description
    --- | ---
    `static ConfigOption<Boolean>` | `LOG_KEY_CONSISTENT`
    `static ConfigOption<Duration>` | `LOG_MAX_READ_TIME`
    `static ConfigOption<Duration>` | `LOG_MAX_WRITE_TIME`
    `static ConfigOption<Duration>` | `LOG_READ_LAG_TIME`
    `static long` | `TIMESLICE_INTERVAL` The time period that is stored under one key in the underlying KCVS.

  - ### Constructor Summary

    Constructors
    Constructor and Description
    `KCVSLog(String name, KCVSLogManager manager, KeyColumnValueStore store, Configuration config)`

  - ### Method Summary

    Methods
    Modifier and Type | Method and Description
    --- | ---
    `Future<Message>` | `add(StaticBuffer content)` ################################### Message Sending ###################################
    `Future<Message>` | `add(StaticBuffer content, StaticBuffer key)` Attempts to add the given content to the log and returns a `Future` for this action.
    `Future<Message>` | `add(StaticBuffer content, StaticBuffer key, ExternalPersistor persistor)`
    `void` | `close()` Closes the log by terminating all threads and waiting for their termination.
    `String` | `getName()` Returns the name of this log
    `StoreTransaction` | `openTx()`
    `void` | `registerReader(ReadMarker readMarker, MessageReader... reader)` ################################### Message Reading ###################################
    `void` | `registerReaders(ReadMarker readMarker, Iterable<MessageReader> readers)` Registers the given readers with this log.
    `boolean` | `unregisterReader(MessageReader reader)` Removes the given reader from the list of registered readers and returns whether this reader was registered in the first place.

    - ### Methods inherited from class java.lang.Object

      `clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait`

- - ### Field Detail

    - #### LOG_MAX_WRITE_TIME

          public static final ConfigOption<Duration> LOG_MAX_WRITE_TIME

    - #### LOG_MAX_READ_TIME

          public static final ConfigOption<Duration> LOG_MAX_READ_TIME

    - #### LOG_READ_LAG_TIME

          public static final ConfigOption<Duration> LOG_READ_LAG_TIME

    - #### LOG_KEY_CONSISTENT

          public static final ConfigOption<Boolean> LOG_KEY_CONSISTENT

    - #### TIMESLICE_INTERVAL

          public static final long TIMESLICE_INTERVAL

      The time period that is stored under one key in the underlying KCVS. This value should NEVER be changed since this will cause backwards incompatibility. This setting is not configurable. If too many messages end up under one key, please configure either 1) the number of buckets or 2) introduce partitioning.
      See Also:
        Constant Field Values

  - ### Constructor Detail

    - #### KCVSLog

          public KCVSLog(String name,
                 KCVSLogManager manager,
                 KeyColumnValueStore store,
                 Configuration config)

  - ### Method Detail

    - #### getName

          public String getName()

      Description copied from interface: `Log`
      Returns the name of this log
      Specified by:
        `getName` in interface `Log`
      Returns:

    - #### close

          public void close()
                     throws BackendException

      Closes the log by terminating all threads and waiting for their termination.
      Specified by:
        `close` in interface `Log`
      Specified by:
        `close` in interface `BackendOperation.TransactionalProvider`
      Throws:
        `BackendException`

    - #### openTx

          public StoreTransaction openTx()
                                  throws BackendException

      Specified by:
        `openTx` in interface `BackendOperation.TransactionalProvider`
      Throws:
        `BackendException`

    - #### add

          public Future<Message> add(StaticBuffer content)

      ################################### Message Sending ###################################
      Specified by:
        `add` in interface `Log`
      Returns:

    - #### add

          public Future<Message> add(StaticBuffer content,
                            StaticBuffer key)

      Description copied from interface: `Log`
      Attempts to add the given content to the log and returns a `Future` for this action. In addition, a key is provided to signal the recipient of the log message in partitioned logging systems. If the log is configured for immediate sending, then any exception encountered during this process is thrown by this method. Otherwise, encountered exceptions are attached to the returned future.
      Specified by:
        `add` in interface `Log`
      Returns:

    - #### add

          public Future<Message> add(StaticBuffer content,
                            StaticBuffer key,
                            ExternalPersistor persistor)

    - #### registerReader

          public void registerReader(ReadMarker readMarker,
                            MessageReader... reader)

      ################################### Message Reading ###################################
      Specified by:
        `registerReader` in interface `Log`
      Parameters:
        `readMarker` - Indicates where to start reading from the log once message readers are registered
        `reader` - The readers to register (all at once)
      See Also:
        `Log.registerReaders(ReadMarker, Iterable)`

    - #### registerReaders

          public void registerReaders(ReadMarker readMarker,
                             Iterable<MessageReader> readers)

      Description copied from interface: `Log`
      Registers the given readers with this log. These readers will be invoked for each newly read message from the log starting at the point identified by the provided `ReadMarker`.

      If no previous readers were registered, invoking this method triggers reader threads to be instantiated. If readers have been previously registered, then the provided `ReadMarker` must be compatible with the previous `ReadMarker` or an exception will be thrown.
      Specified by:
        `registerReaders` in interface `Log`
      Parameters:
        `readMarker` - Indicates where to start reading from the log once message readers are registered
        `readers` - The readers to register (all at once)

    - #### unregisterReader

          public boolean unregisterReader(MessageReader reader)

      Description copied from interface: `Log`
      Removes the given reader from the list of registered readers and returns whether this reader was registered in the first place. Note, that removing the last reader does not stop the reading process. Use `Log.close()` instead.
      Specified by:
        `unregisterReader` in interface `Log`
      Returns:
        true if this MessageReader was registered before and was successfully unregistered, else false

Copyright © 2012–2014. All rights reserved.