com.amazonaws.services.kinesis.connectors

Class KinesisClientLibraryPipelinedRecordProcessor

java.lang.Object

com.amazonaws.services.kinesis.connectors.KinesisClientLibraryPipelinedRecordProcessor

All Implemented Interfaces:

com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessor

public class KinesisClientLibraryPipelinedRecordProcessor
extends Object
implements com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessor

Wraps an IRecordProcessor to pipeline fetching and processing records.

This decouples reading from a stream and processing the records retrieved from the stream. This is useful when either GetRecords or ProcessRecords calls take significant time (e.g. cross-region calls, long-lived tasks, etc). By separating the functionality, the total throughput of the application becomes limited by max(time(GetRecords), time(ProcessRecords)) rather than time(GetRecords) + time(ProcessRecords).

The processRecords method fills a bounded-size, blocking queue with the records to bound the amount of memory required for the record processor. A separate thread consumes the queue and passes batches of records to the wrapped IRecordProcessor.

The IRecordProcessor must checkpoint using the IRecordProcessorCheckpointer.checkpoint(String) method with a specific sequence number. Checkpointing using IRecordProcessorCheckpointer.checkpoint() may only be used in IRecordProcessor.shutdown(IRecordProcessorCheckpointer, ShutdownReason). Using IRecordProcessorCheckpointer.checkpoint() in processRecords will result in an UnsupportedOperationException.

Field Summary

Fields

Modifier and Type	Field and Description
static long	DEFAULT_MAXIMUM_PROCESS_RECORDS_WAIT_TIME_MS Default maximum time to wait for the queue consumer to shutdown (finish ProcessRecords call).
static long	DEFAULT_MAXIMUM_QUEUE_WAIT_TIME_MS Default maximum time to block on the queue waiting for GetRecords result.

Constructor Summary

Constructors

Constructor and Description
KinesisClientLibraryPipelinedRecordProcessor(com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessor recordProcessor, int maxQueueSize) Constructor.
KinesisClientLibraryPipelinedRecordProcessor(com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessor recordProcessor, int maxQueueSize, Long maxQueueWaitTimeMs, Long maxProcessRecordsWaitTimeMs) Constructor.

Method Summary

Modifier and Type	Method and Description
void	initialize(String shardId)
void	processRecords(List<com.amazonaws.services.kinesis.model.Record> records, com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorCheckpointer checkpointer)
void	shutdown(com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorCheckpointer checkpointer, com.amazonaws.services.kinesis.clientlibrary.lib.worker.ShutdownReason reason)

Methods inherited from class java.lang.Object

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

Field Detail

DEFAULT_MAXIMUM_QUEUE_WAIT_TIME_MS

public static final long DEFAULT_MAXIMUM_QUEUE_WAIT_TIME_MS

Default maximum time to block on the queue waiting for GetRecords result.

See Also:

Constant Field Values

DEFAULT_MAXIMUM_PROCESS_RECORDS_WAIT_TIME_MS

public static final long DEFAULT_MAXIMUM_PROCESS_RECORDS_WAIT_TIME_MS

Default maximum time to wait for the queue consumer to shutdown (finish ProcessRecords call).

See Also:

Constant Field Values

Constructor Detail

KinesisClientLibraryPipelinedRecordProcessor

public KinesisClientLibraryPipelinedRecordProcessor(com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessor recordProcessor,
                                                    int maxQueueSize)

Constructor. Default values are used for maximum queue wait time and maximum process records wait time.

Parameters:

recordProcessor - The record processor to wrap

maxQueueSize - The maximum queue size

KinesisClientLibraryPipelinedRecordProcessor

public KinesisClientLibraryPipelinedRecordProcessor(com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessor recordProcessor,
                                                    int maxQueueSize,
                                                    Long maxQueueWaitTimeMs,
                                                    Long maxProcessRecordsWaitTimeMs)

Constructor. If null values are provided for maxQueueWaitTimeMs and/or maxProcessRecordsWaitTimeMs, default values are used.

Parameters:

recordProcessor - The record processor to wrap

maxQueueSize - The maximum queue size

maxQueueWaitTimeMs - Maximum time to block on the queue waiting for GetRecords result in milliseconds

maxProcessRecordsWaitTimeMs - Maximum time to wait for the queue consumer to shutdown (finish ProcessRecords call) in milliseconds

Method Detail

initialize

public void initialize(String shardId)

Specified by:

initialize in interface com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessor

processRecords

public void processRecords(List<com.amazonaws.services.kinesis.model.Record> records,
                           com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorCheckpointer checkpointer)

Specified by:

processRecords in interface com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessor

shutdown

public void shutdown(com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorCheckpointer checkpointer,
                     com.amazonaws.services.kinesis.clientlibrary.lib.worker.ShutdownReason reason)

Specified by:

shutdown in interface com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessor