co.cask.cdap.data.stream.service

Class ConcurrentStreamWriter

java.lang.Object

co.cask.cdap.data.stream.service.ConcurrentStreamWriter

All Implemented Interfaces:

Closeable, AutoCloseable

@ThreadSafe
public final class ConcurrentStreamWriter
extends Object
implements Closeable

Class to support writing to stream with high concurrency. This class supports writing individual stream events as well as appending a new stream file to a stream. For writing individual events to stream, it uses a non-blocking algorithm to batch writes from concurrent threads. The algorithm is like this: When a thread that received a request, for each stream, performs the following:

 1. Constructs a StreamEventData locally and enqueue it to a ConcurrentLinkedQueue.
 2. Use CAS to set an AtomicBoolean flag to true.
 3. If successfully set the flag to true, this thread becomes the writer and proceed to run step 4-7.
 4. Keep polling StreamEventData from the concurrent queue and write to FileWriter with the current timestamp until
    the queue is empty.
 5. Perform a writer flush to make sure all data written are persisted.
 6. Set the state of each StreamEventData that are written to COMPLETED (succeed/failure).
 7. Set the AtomicBoolean flag back to false.
 8. If the StreamEventData enqueued by this thread is NOT COMPLETED, go back to step 2.
 

The spin lock between step 2 to step 8 is necessary as it guarantees events enqueued by all threads would eventually get written and flushed.

Method Summary

Methods

Modifier and Type	Method and Description
void	appendFile(Id.Stream streamId, Location eventFile, Location indexFile, long eventCount, TimestampCloseable timestampCloseable) Appends a new stream file to the given stream.
void	asyncEnqueue(Id.Stream streamId, Map<String,String> headers, ByteBuffer body, Executor executor) Writes an event to the given stream asynchronously.
void	close()
void	enqueue(Id.Stream streamId, Iterator<? extends co.cask.cdap.api.stream.StreamEventData> events) Writes a list of events to the given stream.
void	enqueue(Id.Stream streamId, Map<String,String> headers, ByteBuffer body) Writes an event to the given stream.

Methods inherited from class java.lang.Object

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

Method Detail

enqueue

public void enqueue(Id.Stream streamId,
           Map<String,String> headers,
           ByteBuffer body)
             throws IOException,
                    NotFoundException

Writes an event to the given stream.

Parameters:

streamId - identifier of the stream

headers - header of the event

body - content of the event

Throws:

IOException - if failed to write to stream

IllegalArgumentException - If the stream doesn't exists

NotFoundException

enqueue

public void enqueue(Id.Stream streamId,
           Iterator<? extends co.cask.cdap.api.stream.StreamEventData> events)
             throws IOException,
                    NotFoundException

Writes a list of events to the given stream.

Parameters:

streamId - identifier of the stream

events - list of events to write

Throws:

IOException - if failed to write to stream

IllegalArgumentException - If the stream doesn't exists

NotFoundException

asyncEnqueue

public void asyncEnqueue(Id.Stream streamId,
                Map<String,String> headers,
                ByteBuffer body,
                Executor executor)
                  throws IOException,
                         NotFoundException

Writes an event to the given stream asynchronously. This method returns when the new event is stored to the in-memory event queue, but before persisted.

Parameters:

streamId - identifier of the stream

headers - header of the event

body - content of the event

executor - The executor for performing the async write flush operation

Throws:

IOException - if fails to get stream information

IllegalArgumentException - If the stream doesn't exists

NotFoundException

appendFile

public void appendFile(Id.Stream streamId,
              Location eventFile,
              Location indexFile,
              long eventCount,
              TimestampCloseable timestampCloseable)
                throws IOException,
                       NotFoundException

Appends a new stream file to the given stream.

Parameters:

streamId - identifier of the stream

eventFile - location to the new stream data file

indexFile - location to the new stream index file

eventCount - number of events in the given stream file

timestampCloseable - a TimestampCloseable to close and return the stream file close timestamp

Throws:

IOException - if failed to append the new stream file

NotFoundException

close

public void close()
           throws IOException

Specified by:

close in interface Closeable

Specified by:

close in interface AutoCloseable

Throws:

IOException