co.cask.cdap.data2.dataset2

Class DynamicDatasetCache

java.lang.Object

co.cask.cdap.data2.dataset2.DynamicDatasetCache

All Implemented Interfaces:

co.cask.cdap.api.data.DatasetContext, com.google.common.base.Supplier<co.cask.tephra.TransactionContext>

Direct Known Subclasses:

MultiThreadDatasetCache, SingleThreadDatasetCache

public abstract class DynamicDatasetCache
extends Object
implements co.cask.cdap.api.data.DatasetContext, com.google.common.base.Supplier<co.cask.tephra.TransactionContext>

Implementation of DatasetContext that allows to dynamically load datasets into a started TransactionContext. Datasets acquired from this context are distinct from any Datasets instantiated outside this class. Datasets are cached, such that repeated calls to (@link #getDataset()} for the same dataset and arguments return the same instance. The cache also maintains a transaction context and adds all acquired datasets to that context, so that they participate in the transactions executed with that context. If a dataset is dismissed during the course of a transaction, then this context delays the dismissal until the transaction is complete. Optionally, this cache can have a set of static datasets that are added to every transaction context created by the cache. Static datasets cannot be dismissed. Also, transaction-aware "datasets" that were not created by this DynamicDatasetCache, can be added to the transaction context. This is useful for transaction-aware's that do not implement a Dataset (such as queue consumers etc.).

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
protected static class	DynamicDatasetCache.DatasetCacheKey A key used by implementations of DynamicDatasetCache to cache Datasets.

Field Summary

Fields

Modifier and Type	Field and Description
protected SystemDatasetInstantiator	instantiator
protected MetricsContext	metricsContext
protected NamespaceId	namespace
protected Map<String,String>	runtimeArguments
protected co.cask.tephra.TransactionSystemClient	txClient

Constructor Summary

Constructors

Constructor and Description
DynamicDatasetCache(SystemDatasetInstantiator instantiator, co.cask.tephra.TransactionSystemClient txClient, NamespaceId namespace, Map<String,String> runtimeArguments, MetricsContext metricsContext) Create a dynamic dataset factory.

Method Summary

Methods

Modifier and Type	Method and Description
abstract void	addExtraTransactionAware(co.cask.tephra.TransactionAware txAware) Add an extra transaction aware to the static datasets.
void	close() Close and dismiss all datasets that were obtained through this factory, and destroy the factory.
abstract void	dismissTransactionContext() Dismiss the current transaction context.
co.cask.tephra.TransactionContext	get()
protected abstract <T extends co.cask.cdap.api.dataset.Dataset> T	getDataset(DynamicDatasetCache.DatasetCacheKey key, boolean bypass) To be implemented by subclasses.
<T extends co.cask.cdap.api.dataset.Dataset> T	getDataset(String name)
<T extends co.cask.cdap.api.dataset.Dataset> T	getDataset(String name, Map<String,String> arguments)
<T extends co.cask.cdap.api.dataset.Dataset> T	getDataset(String name, Map<String,String> arguments, boolean bypass) Instantiate a dataset, allowing to bypass the cache.
abstract Iterable<co.cask.tephra.TransactionAware>	getStaticTransactionAwares()
abstract Iterable<co.cask.tephra.TransactionAware>	getTransactionAwares()
abstract void	invalidate() Close and dismiss all datasets that were obtained through this factory.
abstract co.cask.tephra.TransactionContext	newTransactionContext() Return a new transaction context for the current thread.
void	releaseDataset(co.cask.cdap.api.dataset.Dataset dataset)
abstract void	removeExtraTransactionAware(co.cask.tephra.TransactionAware txAware) Remove a transaction-aware that was added via addExtraTransactionAware(TransactionAware).

Methods inherited from class java.lang.Object

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

Methods inherited from interface co.cask.cdap.api.data.DatasetContext

discardDataset

Field Detail

instantiator

protected final SystemDatasetInstantiator instantiator

txClient

protected final co.cask.tephra.TransactionSystemClient txClient

namespace

protected final NamespaceId namespace

runtimeArguments

protected final Map<String,String> runtimeArguments

metricsContext

protected final MetricsContext metricsContext

Constructor Detail

DynamicDatasetCache

public DynamicDatasetCache(SystemDatasetInstantiator instantiator,
                   co.cask.tephra.TransactionSystemClient txClient,
                   NamespaceId namespace,
                   Map<String,String> runtimeArguments,
                   @Nullable
                   MetricsContext metricsContext)

Create a dynamic dataset factory.

Parameters:

txClient - the transaction system client to use for new transaction contexts

namespace - the NamespaceId in which all datasets are instantiated

runtimeArguments - all runtime arguments that are available to datasets in the context. Runtime arguments are expected to be scoped so that arguments for one dataset do not override arguments of other datasets.

metricsContext - if non-null, this context is used as the context for dataset metrics, with an additional tag for the dataset name.

Method Detail

getDataset

public final <T extends co.cask.cdap.api.dataset.Dataset> T getDataset(String name)
                                                            throws co.cask.cdap.api.data.DatasetInstantiationException

Specified by:

getDataset in interface co.cask.cdap.api.data.DatasetContext

Throws:

co.cask.cdap.api.data.DatasetInstantiationException

getDataset

public final <T extends co.cask.cdap.api.dataset.Dataset> T getDataset(String name,
                                                        Map<String,String> arguments)
                                                            throws co.cask.cdap.api.data.DatasetInstantiationException

Specified by:

getDataset in interface co.cask.cdap.api.data.DatasetContext

Throws:

co.cask.cdap.api.data.DatasetInstantiationException

getDataset

public final <T extends co.cask.cdap.api.dataset.Dataset> T getDataset(String name,
                                                        Map<String,String> arguments,
                                                        boolean bypass)
                                                            throws co.cask.cdap.api.data.DatasetInstantiationException

Instantiate a dataset, allowing to bypass the cache. This means that the dataset will not be added to the in-progress transactions, and it will also not be closed when the cache is closed.

Type Parameters:

T - the type of the dataset

Parameters:

name - the name of the dataset

arguments - arguments for the dataset

bypass - whether to bypass the cache

Throws:

co.cask.cdap.api.data.DatasetInstantiationException

releaseDataset

public void releaseDataset(co.cask.cdap.api.dataset.Dataset dataset)

Specified by:

releaseDataset in interface co.cask.cdap.api.data.DatasetContext

getDataset

protected abstract <T extends co.cask.cdap.api.dataset.Dataset> T getDataset(DynamicDatasetCache.DatasetCacheKey key,
                                                        boolean bypass)
                                                                  throws co.cask.cdap.api.data.DatasetInstantiationException

To be implemented by subclasses.

Parameters:

bypass - if true, bypass the dataset cache, and do not add this to the transaction.

Throws:

co.cask.cdap.api.data.DatasetInstantiationException

newTransactionContext

public abstract co.cask.tephra.TransactionContext newTransactionContext()

Return a new transaction context for the current thread. All transaction-aware static datasets and all extra transaction-awares are added to this transaction initially. Also, any transaction-aware that was previously dynamically acquired, and that has not been garbage-collected, is added to the transaction. Any transaction-aware datasets that will subsequently be obtained via (@link #getDataset()) will then also be added to this transaction context and thus participate in its transaction. These datasets can also be retrieved using getTransactionAwares().

Returns:

a new transaction context

dismissTransactionContext

public abstract void dismissTransactionContext()

Dismiss the current transaction context. This releases the references to the context's transaction-aware datasets so that they can be collected by the garbage collector (if no one else is holding a reference to them). The static datasets and the extra transaction-awares, however, will not be made available to garbage collection, and will participate in the next transaction (created by newTransactionContext()).

get

public co.cask.tephra.TransactionContext get()

Specified by:

get in interface com.google.common.base.Supplier<co.cask.tephra.TransactionContext>

getStaticTransactionAwares

public abstract Iterable<co.cask.tephra.TransactionAware> getStaticTransactionAwares()

Returns:

the static datasets that are transaction-aware. This is the same independent of whether a transaction context was started using newTransactionContext().

getTransactionAwares

public abstract Iterable<co.cask.tephra.TransactionAware> getTransactionAwares()

Returns:

the transaction-aware datasets that participate in the current transaction. If newTransactionContext() has not been called (or dismissTransactionContext() has been called), then there is no transaction and this will return an empty iterable.

addExtraTransactionAware

public abstract void addExtraTransactionAware(co.cask.tephra.TransactionAware txAware)

Add an extra transaction aware to the static datasets. This is a transaction aware that is not instantiated through this factory, but needs to participate in every transaction. Note that if a transaction is in progress, then this transaction aware will join that transaction.

removeExtraTransactionAware

public abstract void removeExtraTransactionAware(co.cask.tephra.TransactionAware txAware)

Remove a transaction-aware that was added via addExtraTransactionAware(TransactionAware). Note that if a transaction is in progress, then this transaction aware will leave that transaction.

close

public void close()

Close and dismiss all datasets that were obtained through this factory, and destroy the factory. If an extra transaction-awares were added to this cache (and not removed), then they will also be closed.

invalidate

public abstract void invalidate()

Close and dismiss all datasets that were obtained through this factory. This can be used to ensure that all resources held by datasets are released, even though the factory may be still be used for subsequent execution.