co.cask.cdap.data2.transaction.stream

Class AbstractStreamFileConsumer

java.lang.Object

co.cask.cdap.data2.transaction.stream.AbstractStreamFileConsumer

All Implemented Interfaces:

StreamConsumer, co.cask.tephra.TransactionAware, Closeable, AutoCloseable

Direct Known Subclasses:

HBaseStreamFileConsumer, LevelDBStreamFileConsumer

@NotThreadSafe
public abstract class AbstractStreamFileConsumer
extends Object
implements StreamConsumer

A StreamConsumer that read events from stream file and uses a table to store consumer states.

State table ROW key schema:

row_key = <group_id> <stream_file_offset>
   group_id = 8 bytes consumer group id
   stream_file_offset = <partition_start> <partition_end> <name_prefix> <sequence_id> <offset>
   partition_start = 8 bytes timestamp of partition start time
   partition_end = 8 bytes timestamp of partition end time
   name_prefix = DataOutput UTF-8 output.
   sequence_id = 4 bytes stream file sequence id
   offset = 8 bytes offset inside the stream file
 

The state table has single column (QueueEntryRow.COLUMN_FAMILY:QueueEntryRow.STATE_COLUMN_PREFIX) to store state. The state value:

state_value = <write_pointer> <instance_id> <state>
   write_pointer = 8 bytes Transaction write point of the consumer who update this state.
   instance_id = 4 bytes Instance id of the consumer who update this state.
   state = ConsumerEntryState.getState(), either CLAIMED or PROCESSED
 

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
protected static interface	AbstractStreamFileConsumer.StateScanner Scanner for scanning state table.

Field Summary

Fields

Modifier and Type	Field and Description
protected static int	MAX_SCAN_ROWS
protected byte[]	stateColumnName

Constructor Summary

Constructors

Modifier	Constructor and Description
protected	AbstractStreamFileConsumer(CConfiguration cConf, StreamConfig streamConfig, ConsumerConfig consumerConfig, FileReader<StreamEventOffset,Iterable<StreamFileOffset>> reader, StreamConsumerStateStore consumerStateStore, StreamConsumerState beginConsumerState, ReadFilter extraFilter)

Method Summary

Methods

Modifier and Type	Method and Description
protected abstract boolean	claimFifoEntry(byte[] row, byte[] value, byte[] oldValue)
void	close()
boolean	commitTx()
protected void	doClose()
ConsumerConfig	getConsumerConfig()
Id.Stream	getStreamId()
String	getTransactionAwareName()
Collection<byte[]>	getTxChanges()
DequeueResult<co.cask.cdap.api.flow.flowlet.StreamEvent>	poll(int maxEvents, long timeout, TimeUnit timeoutUnit) Retrieves up to maxEvents of StreamEvent from the stream.
void	postTxCommit()
boolean	rollbackTx()
protected abstract AbstractStreamFileConsumer.StateScanner	scanStates(byte[] startRow, byte[] endRow)
void	startTx(co.cask.tephra.Transaction tx)
String	toString()
protected abstract void	undoState(Iterable<byte[]> rows, int size)
protected abstract void	updateState(Iterable<byte[]> rows, int size, byte[] value)
void	updateTx(co.cask.tephra.Transaction tx)

Methods inherited from class java.lang.Object

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

Field Detail

MAX_SCAN_ROWS

protected static final int MAX_SCAN_ROWS

See Also:

Constant Field Values

stateColumnName

protected final byte[] stateColumnName

Constructor Detail

AbstractStreamFileConsumer

protected AbstractStreamFileConsumer(CConfiguration cConf,
                          StreamConfig streamConfig,
                          ConsumerConfig consumerConfig,
                          FileReader<StreamEventOffset,Iterable<StreamFileOffset>> reader,
                          StreamConsumerStateStore consumerStateStore,
                          StreamConsumerState beginConsumerState,
                          @Nullable
                          ReadFilter extraFilter)

Parameters:

streamConfig - Stream configuration.

consumerConfig - Consumer configuration.

reader - For reading stream events. This class is responsible for closing the reader.

consumerStateStore - The state store for saving consumer state

beginConsumerState - Consumer state to begin with.

extraFilter - Extra ReadFilter that is ANDed with default read filter and applied first.

Method Detail

doClose

protected void doClose()
                throws IOException

Throws:

IOException

claimFifoEntry

protected abstract boolean claimFifoEntry(byte[] row,
                     byte[] value,
                     byte[] oldValue)
                                   throws IOException

Throws:

IOException

updateState

protected abstract void updateState(Iterable<byte[]> rows,
               int size,
               byte[] value)
                             throws IOException

Throws:

IOException

undoState

protected abstract void undoState(Iterable<byte[]> rows,
             int size)
                           throws IOException

Throws:

IOException

scanStates

protected abstract AbstractStreamFileConsumer.StateScanner scanStates(byte[] startRow,
                                                 byte[] endRow)
                                                               throws IOException

Throws:

IOException

getStreamId

public final Id.Stream getStreamId()

Specified by:

getStreamId in interface StreamConsumer

Returns:

Id of the stream this consumer is consuming.

getConsumerConfig

public final ConsumerConfig getConsumerConfig()

Specified by:

getConsumerConfig in interface StreamConsumer

Returns:

Configuration of this consumer.

poll

public final DequeueResult<co.cask.cdap.api.flow.flowlet.StreamEvent> poll(int maxEvents,
                                                            long timeout,
                                                            TimeUnit timeoutUnit)
                                                                    throws IOException,
                                                                           InterruptedException

Description copied from interface: StreamConsumer

Retrieves up to maxEvents of StreamEvent from the stream.

Specified by:

poll in interface StreamConsumer

Parameters:

maxEvents - Maximum number of events to retrieve

timeout - Maximum of time to spend on trying to read up to maxEvents

timeoutUnit - Unit for the timeout

Returns:

An instance of DequeueResult which carries StreamEvents inside.

Throws:

IOException - If there is error while reading events

InterruptedException - If interrupted while waiting

close

public final void close()
                 throws IOException

Specified by:

close in interface Closeable

Specified by:

close in interface AutoCloseable

Throws:

IOException

startTx

public final void startTx(co.cask.tephra.Transaction tx)

Specified by:

startTx in interface co.cask.tephra.TransactionAware

updateTx

public final void updateTx(co.cask.tephra.Transaction tx)

Specified by:

updateTx in interface co.cask.tephra.TransactionAware

getTxChanges

public final Collection<byte[]> getTxChanges()

Specified by:

getTxChanges in interface co.cask.tephra.TransactionAware

commitTx

public final boolean commitTx()
                       throws Exception

Specified by:

commitTx in interface co.cask.tephra.TransactionAware

Throws:

Exception

postTxCommit

public void postTxCommit()

Specified by:

postTxCommit in interface co.cask.tephra.TransactionAware

rollbackTx

public boolean rollbackTx()
                   throws Exception

Specified by:

rollbackTx in interface co.cask.tephra.TransactionAware

Throws:

Exception

getTransactionAwareName

public String getTransactionAwareName()

Specified by:

getTransactionAwareName in interface co.cask.tephra.TransactionAware

toString

public String toString()

Overrides:

toString in class Object